From 1402152540fa584ea856ddd24a02370286e4971f Mon Sep 17 00:00:00 2001 From: king-aj-the-first Date: Sun, 26 Jul 2026 16:23:36 +0100 Subject: [PATCH] Add channel subscriptions, counts, creators, and DB query indexes --- .../contracts/hello-world/src/base/channel.rs | 176 ++++++++++++++ .../hello-world/src/channel_logic.rs | 227 ++++++++++++++++++ contract/contracts/hello-world/src/lib.rs | 60 +++++ .../src/tests/channel_subscription_test.rs | 188 +++++++++++++++ docs/BATCH_SUBSCRIBE_GAS.md | 48 ++++ docs/DATABASE_QUERY_PERFORMANCE.md | 101 ++++++++ .../src/database/query-performance.test.ts | 78 ++++++ listener/src/database/schema.sql | 49 ++++ .../002-query-performance-indexes.ts | 74 ++++++ 9 files changed, 1001 insertions(+) create mode 100644 contract/contracts/hello-world/src/base/channel.rs create mode 100644 contract/contracts/hello-world/src/channel_logic.rs create mode 100644 contract/contracts/hello-world/src/tests/channel_subscription_test.rs create mode 100644 docs/BATCH_SUBSCRIBE_GAS.md create mode 100644 docs/DATABASE_QUERY_PERFORMANCE.md create mode 100644 listener/src/database/query-performance.test.ts create mode 100644 listener/src/migrations/002-query-performance-indexes.ts diff --git a/contract/contracts/hello-world/src/base/channel.rs b/contract/contracts/hello-world/src/base/channel.rs new file mode 100644 index 0000000..6257ef0 --- /dev/null +++ b/contract/contracts/hello-world/src/base/channel.rs @@ -0,0 +1,176 @@ +/// Notification channel types and storage helpers. +/// +/// Channels are named notification streams that users can subscribe to. +/// Each channel records its original creator and maintains an active +/// subscriber count for efficient read-only queries. +use soroban_sdk::{contractevent, contracttype, Address, BytesN, Env, String, Vec}; + +use crate::base::events::{NotificationCategory, NotificationPriority}; + +// ============================================================================ +// Types +// ============================================================================ + +/// On-chain notification channel metadata. +#[contracttype] +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct NotificationChannel { + /// Unique channel identifier. + pub id: BytesN<32>, + /// Wallet that originally created the channel. + pub creator: Address, + /// Human-readable channel name. + pub name: String, + /// Number of currently active subscribers. + pub subscriber_count: u32, + /// Whether the channel accepts new subscriptions. + pub is_active: bool, + /// Ledger timestamp when the channel was created. + pub created_at: u64, +} + +/// Result summary returned by batch subscription. +#[contracttype] +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BatchSubscribeResult { + /// Number of channels successfully subscribed. + pub succeeded: u32, + /// Number of channels skipped due to errors (not found, inactive, already subscribed). + pub failed: u32, + /// Channel ids that were successfully subscribed in this call. + pub subscribed_ids: Vec>, +} + +// ============================================================================ +// Storage keys +// ============================================================================ + +#[contracttype] +#[derive(Clone)] +pub enum ChannelDataKey { + /// Channel metadata keyed by channel id. + Channel(BytesN<32>), + /// Ordered list of all channel ids. + AllChannels, + /// Active subscription flag: (channel_id, subscriber) -> bool. + Subscription(BytesN<32>, Address), + /// List of subscriber addresses for a channel (for enumeration). + ChannelSubscribers(BytesN<32>), +} + +/// Maximum channels allowed in a single `batch_subscribe` call. +pub const MAX_BATCH_SUBSCRIBE: u32 = 50; +/// Maximum length for a channel name. +pub const MAX_CHANNEL_NAME_LENGTH: u32 = 100; + +// ============================================================================ +// Storage helpers +// ============================================================================ + +pub fn load_channel(env: &Env, id: &BytesN<32>) -> Option { + env.storage() + .persistent() + .get(&ChannelDataKey::Channel(id.clone())) +} + +pub fn save_channel(env: &Env, channel: &NotificationChannel) { + env.storage() + .persistent() + .set(&ChannelDataKey::Channel(channel.id.clone()), channel); +} + +pub fn is_subscribed(env: &Env, channel_id: &BytesN<32>, subscriber: &Address) -> bool { + env.storage() + .persistent() + .get(&ChannelDataKey::Subscription(channel_id.clone(), subscriber.clone())) + .unwrap_or(false) +} + +pub fn set_subscribed(env: &Env, channel_id: &BytesN<32>, subscriber: &Address, value: bool) { + let key = ChannelDataKey::Subscription(channel_id.clone(), subscriber.clone()); + if value { + env.storage().persistent().set(&key, &true); + } else { + env.storage().persistent().remove(&key); + } +} + +pub fn load_subscribers(env: &Env, channel_id: &BytesN<32>) -> Vec
{ + env.storage() + .persistent() + .get(&ChannelDataKey::ChannelSubscribers(channel_id.clone())) + .unwrap_or(Vec::new(env)) +} + +pub fn save_subscribers(env: &Env, channel_id: &BytesN<32>, subscribers: &Vec
) { + env.storage() + .persistent() + .set( + &ChannelDataKey::ChannelSubscribers(channel_id.clone()), + subscribers, + ); +} + +pub fn push_all_channel_id(env: &Env, id: &BytesN<32>) { + let key = ChannelDataKey::AllChannels; + let mut all: Vec> = env.storage().persistent().get(&key).unwrap_or(Vec::new(env)); + all.push_back(id.clone()); + env.storage().persistent().set(&key, &all); +} + +// ============================================================================ +// Events +// ============================================================================ + +/// Emitted when a notification channel is created. +#[contractevent(data_format = "single-value")] +#[derive(Clone)] +pub struct ChannelCreated { + #[topic] + pub creator: Address, + #[topic] + pub category: NotificationCategory, + #[topic] + pub priority: NotificationPriority, + pub channel_id: BytesN<32>, +} + +/// Emitted when a user successfully subscribes to a channel. +#[contractevent(data_format = "single-value")] +#[derive(Clone)] +pub struct ChannelSubscribed { + #[topic] + pub channel_id: BytesN<32>, + #[topic] + pub subscriber: Address, + #[topic] + pub category: NotificationCategory, + pub subscriber_count: u32, +} + +/// Emitted when a user unsubscribes from a channel. +#[contractevent(data_format = "single-value")] +#[derive(Clone)] +pub struct ChannelUnsubscribed { + #[topic] + pub channel_id: BytesN<32>, + #[topic] + pub subscriber: Address, + #[topic] + pub category: NotificationCategory, + pub subscriber_count: u32, +} + +/// Summary event for a batch subscribe call. +#[contractevent] +#[derive(Clone)] +pub struct BatchSubscribeCompleted { + #[topic] + pub subscriber: Address, + #[topic] + pub category: NotificationCategory, + #[topic] + pub priority: NotificationPriority, + pub succeeded: u32, + pub failed: u32, +} diff --git a/contract/contracts/hello-world/src/channel_logic.rs b/contract/contracts/hello-world/src/channel_logic.rs new file mode 100644 index 0000000..551f024 --- /dev/null +++ b/contract/contracts/hello-world/src/channel_logic.rs @@ -0,0 +1,227 @@ +/// Notification channel creation, subscription, and subscriber-count views. +use crate::base::channel::{ + is_subscribed, load_channel, load_subscribers, push_all_channel_id, save_channel, + save_subscribers, set_subscribed, BatchSubscribeCompleted, BatchSubscribeResult, + ChannelCreated, ChannelSubscribed, ChannelUnsubscribed, NotificationChannel, + MAX_BATCH_SUBSCRIBE, MAX_CHANNEL_NAME_LENGTH, +}; +use crate::base::errors::Error; +use crate::base::events::{NotificationCategory, NotificationPriority}; +use soroban_sdk::{Address, BytesN, Env, String, Vec}; + +fn require_not_paused(env: &Env) -> Result<(), Error> { + if crate::autoshare_logic::get_paused_status(env) { + return Err(Error::ContractPaused); + } + Ok(()) +} + +/// Creates a new notification channel owned by `creator`. +/// +/// Stores the creator address permanently so it can be queried later. +/// Emits a [`ChannelCreated`] event. +pub fn create_channel( + env: Env, + id: BytesN<32>, + name: String, + creator: Address, +) -> Result<(), Error> { + creator.require_auth(); + require_not_paused(&env)?; + + if name.is_empty() { + return Err(Error::InvalidInput); + } + if name.len() > MAX_CHANNEL_NAME_LENGTH { + return Err(Error::NameTooLong); + } + + if load_channel(&env, &id).is_some() { + return Err(Error::AlreadyExists); + } + + let channel = NotificationChannel { + id: id.clone(), + creator: creator.clone(), + name, + subscriber_count: 0, + is_active: true, + created_at: env.ledger().timestamp(), + }; + save_channel(&env, &channel); + push_all_channel_id(&env, &id); + + ChannelCreated { + creator, + category: NotificationCategory::Notification, + priority: NotificationPriority::Medium, + channel_id: id, + } + .publish(&env); + + Ok(()) +} + +/// Returns channel metadata including creator and subscriber count. +pub fn get_channel(env: Env, id: BytesN<32>) -> Result { + load_channel(&env, &id).ok_or(Error::NotFound) +} + +/// Returns the wallet address that originally created the channel. +pub fn get_channel_creator(env: Env, id: BytesN<32>) -> Result { + Ok(load_channel(&env, &id).ok_or(Error::NotFound)?.creator) +} + +/// Read-only view of the active subscriber count for a channel. +pub fn get_subscriber_count(env: Env, id: BytesN<32>) -> Result { + Ok(load_channel(&env, &id) + .ok_or(Error::NotFound)? + .subscriber_count) +} + +/// Returns whether `subscriber` is currently subscribed to the channel. +pub fn is_channel_subscriber(env: Env, id: BytesN<32>, subscriber: Address) -> bool { + is_subscribed(&env, &id, &subscriber) +} + +/// Subscribe a single address to a channel. +pub fn subscribe(env: Env, channel_id: BytesN<32>, subscriber: Address) -> Result<(), Error> { + subscriber.require_auth(); + require_not_paused(&env)?; + subscribe_one(&env, &channel_id, &subscriber)?; + Ok(()) +} + +/// Unsubscribe from a channel. No-ops the count if not currently subscribed. +pub fn unsubscribe(env: Env, channel_id: BytesN<32>, subscriber: Address) -> Result<(), Error> { + subscriber.require_auth(); + require_not_paused(&env)?; + + let mut channel = load_channel(&env, &channel_id).ok_or(Error::NotFound)?; + + if !is_subscribed(&env, &channel_id, &subscriber) { + return Err(Error::NotFound); + } + + set_subscribed(&env, &channel_id, &subscriber, false); + + let mut subscribers = load_subscribers(&env, &channel_id); + let mut next = Vec::new(&env); + for addr in subscribers.iter() { + if addr != subscriber { + next.push_back(addr); + } + } + save_subscribers(&env, &channel_id, &next); + + channel.subscriber_count = channel.subscriber_count.saturating_sub(1); + save_channel(&env, &channel); + + ChannelUnsubscribed { + channel_id, + subscriber, + category: NotificationCategory::Notification, + subscriber_count: channel.subscriber_count, + } + .publish(&env); + + Ok(()) +} + +/// Subscribe to multiple channels in a single transaction. +/// +/// Failed individual subscriptions (missing channel, inactive, already +/// subscribed) are skipped and do not roll back successful ones — state for +/// failed entries is left unchanged. Structural errors (empty batch, too large, +/// paused contract) abort the whole call before any mutation. +/// +/// # Gas notes +/// See `docs/BATCH_SUBSCRIBE_GAS.md`. Batching N subscriptions into one +/// transaction avoids N separate transaction base fees and repeated auth +/// overhead. Per-channel storage writes still scale linearly with N. +pub fn batch_subscribe( + env: Env, + channel_ids: Vec>, + subscriber: Address, +) -> Result { + subscriber.require_auth(); + require_not_paused(&env)?; + + let count = channel_ids.len(); + if count == 0 { + return Err(Error::InvalidInput); + } + if count > MAX_BATCH_SUBSCRIBE { + return Err(Error::BatchTooLarge); + } + + let mut succeeded = 0u32; + let mut failed = 0u32; + let mut subscribed_ids = Vec::new(&env); + + for i in 0..count { + let channel_id = channel_ids.get(i).unwrap(); + match subscribe_one(&env, &channel_id, &subscriber) { + Ok(()) => { + succeeded += 1; + subscribed_ids.push_back(channel_id); + } + Err(_) => { + failed += 1; + } + } + } + + BatchSubscribeCompleted { + subscriber: subscriber.clone(), + category: NotificationCategory::Notification, + priority: NotificationPriority::Medium, + succeeded, + failed, + } + .publish(&env); + + Ok(BatchSubscribeResult { + succeeded, + failed, + subscribed_ids, + }) +} + +fn subscribe_one( + env: &Env, + channel_id: &BytesN<32>, + subscriber: &Address, +) -> Result<(), Error> { + let mut channel = load_channel(env, channel_id).ok_or(Error::NotFound)?; + + if !channel.is_active { + return Err(Error::GroupInactive); + } + + if is_subscribed(env, channel_id, subscriber) { + return Err(Error::AlreadyExists); + } + + set_subscribed(env, channel_id, subscriber, true); + + let mut subscribers = load_subscribers(env, channel_id); + subscribers.push_back(subscriber.clone()); + save_subscribers(env, channel_id, &subscribers); + + channel.subscriber_count = channel + .subscriber_count + .checked_add(1) + .ok_or(Error::InvalidInput)?; + save_channel(env, &channel); + + ChannelSubscribed { + channel_id: channel_id.clone(), + subscriber: subscriber.clone(), + category: NotificationCategory::Notification, + subscriber_count: channel.subscriber_count, + } + .publish(env); + + Ok(()) +} diff --git a/contract/contracts/hello-world/src/lib.rs b/contract/contracts/hello-world/src/lib.rs index f834d9b..55ac8fd 100644 --- a/contract/contracts/hello-world/src/lib.rs +++ b/contract/contracts/hello-world/src/lib.rs @@ -3,6 +3,7 @@ use soroban_sdk::{contract, contractimpl, Address, BytesN, Env, String, Vec}; // 1. Declare the foundational modules (Requirement: Modular Structure) pub mod base { + pub mod channel; pub mod errors; pub mod events; pub mod metadata_validation; @@ -17,6 +18,7 @@ pub mod interfaces { // 2. Declare the main logic files where the functions are implemented mod autoshare_logic; +mod channel_logic; mod preferences_logic; mod reputation_logic; @@ -454,6 +456,8 @@ impl AutoShareContract { /// The notification must exist, not already be revoked or expired, and not yet be delivered. pub fn recall_notification(env: Env, notification_id: BytesN<32>, caller: Address) { autoshare_logic::recall_notification(env, notification_id, caller).unwrap(); + } + /// Emits a `BatchProcessingCompleted` event for off-chain listeners. pub fn emit_batch_completed(env: Env, batch_id: BytesN<32>, processed_count: u32) { autoshare_logic::emit_batch_completed(env, batch_id, processed_count).unwrap(); @@ -527,6 +531,8 @@ impl AutoShareContract { /// Acknowledges multiple scheduled notifications in a single batch. pub fn acknowledge_notifications(env: Env, caller: Address, notification_ids: Vec>) { autoshare_logic::acknowledge_notifications(env, caller, notification_ids).unwrap(); + } + /// Extends the expiration period of a scheduled notification by `extension_seconds`. /// /// Only the notification creator or the contract admin can extend it. @@ -642,6 +648,57 @@ impl AutoShareContract { pub fn record_notification_access(env: Env, notification_id: BytesN<32>, accessor: Address) { autoshare_logic::record_notification_access(env, notification_id, accessor).unwrap(); } + + // ============================================================================ + // Notification Channel Subscriptions + // ============================================================================ + + /// Creates a notification channel and permanently stores the creator address. + pub fn create_channel(env: Env, id: BytesN<32>, name: String, creator: Address) { + channel_logic::create_channel(env, id, name, creator).unwrap(); + } + + /// Returns full channel metadata (creator, name, subscriber_count, etc.). + pub fn get_channel(env: Env, id: BytesN<32>) -> base::channel::NotificationChannel { + channel_logic::get_channel(env, id).unwrap() + } + + /// Returns the wallet address that originally created the channel. + pub fn get_channel_creator(env: Env, id: BytesN<32>) -> Address { + channel_logic::get_channel_creator(env, id).unwrap() + } + + /// Read-only view of the active subscriber count for a channel. + pub fn get_subscriber_count(env: Env, id: BytesN<32>) -> u32 { + channel_logic::get_subscriber_count(env, id).unwrap() + } + + /// Returns whether `subscriber` is currently subscribed to the channel. + pub fn is_channel_subscriber(env: Env, id: BytesN<32>, subscriber: Address) -> bool { + channel_logic::is_channel_subscriber(env, id, subscriber) + } + + /// Subscribe to a single notification channel. + pub fn subscribe(env: Env, channel_id: BytesN<32>, subscriber: Address) { + channel_logic::subscribe(env, channel_id, subscriber).unwrap(); + } + + /// Unsubscribe from a notification channel. + pub fn unsubscribe(env: Env, channel_id: BytesN<32>, subscriber: Address) { + channel_logic::unsubscribe(env, channel_id, subscriber).unwrap(); + } + + /// Subscribe to multiple channels in one transaction. + /// + /// Individual failures are skipped without corrupting successful subscriptions. + /// See `docs/BATCH_SUBSCRIBE_GAS.md` for gas documentation. + pub fn batch_subscribe( + env: Env, + channel_ids: Vec>, + subscriber: Address, + ) -> base::channel::BatchSubscribeResult { + channel_logic::batch_subscribe(env, channel_ids, subscriber).unwrap() + } } #[cfg(test)] @@ -725,4 +782,7 @@ mod tests { #[path = "../tests/subscription_cancellation_test.rs"] mod subscription_cancellation_test; + + #[path = "../tests/channel_subscription_test.rs"] + mod channel_subscription_test; } diff --git a/contract/contracts/hello-world/src/tests/channel_subscription_test.rs b/contract/contracts/hello-world/src/tests/channel_subscription_test.rs new file mode 100644 index 0000000..10c36a1 --- /dev/null +++ b/contract/contracts/hello-world/src/tests/channel_subscription_test.rs @@ -0,0 +1,188 @@ +//! Tests for notification channel subscriptions: +//! - subscriber count view +//! - batch subscribe (partial failure safety) +//! - channel creator storage / query + +use crate::base::errors::Error; +use crate::test_utils::setup_test_env; +use crate::AutoShareContractClient; + +use soroban_sdk::{BytesN, String, Vec}; + +fn make_id(env: &soroban_sdk::Env, tag: u8) -> BytesN<32> { + let mut bytes = [0u8; 32]; + bytes[0] = tag; + BytesN::from_array(env, &bytes) +} + +fn make_name(env: &soroban_sdk::Env, label: &str) -> String { + String::from_str(env, label) +} + +// ============================================================================ +// Task 3 — Channel creator information +// ============================================================================ + +#[test] +fn test_create_channel_stores_creator() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let creator = test_env.users.get(0).unwrap().clone(); + let id = make_id(&test_env.env, 1); + + client.create_channel(&id, &make_name(&test_env.env, "alerts"), &creator); + + let channel = client.get_channel(&id); + assert_eq!(channel.creator, creator); + assert_eq!(channel.subscriber_count, 0); + assert!(channel.is_active); + assert_eq!(client.get_channel_creator(&id), creator); +} + +#[test] +fn test_get_channel_creator_not_found() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let id = make_id(&test_env.env, 99); + + let result = client.try_get_channel_creator(&id); + assert_eq!(result, Err(Ok(Error::NotFound))); +} + +// ============================================================================ +// Task 1 — Subscription count view +// ============================================================================ + +#[test] +fn test_subscriber_count_updates_on_subscribe_and_unsubscribe() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let creator = test_env.users.get(0).unwrap().clone(); + let user_a = test_env.users.get(1).unwrap().clone(); + let user_b = test_env.users.get(2).unwrap().clone(); + let id = make_id(&test_env.env, 2); + + client.create_channel(&id, &make_name(&test_env.env, "news"), &creator); + assert_eq!(client.get_subscriber_count(&id), 0); + + client.subscribe(&id, &user_a); + assert_eq!(client.get_subscriber_count(&id), 1); + assert!(client.is_channel_subscriber(&id, &user_a)); + + client.subscribe(&id, &user_b); + assert_eq!(client.get_subscriber_count(&id), 2); + + client.unsubscribe(&id, &user_a); + assert_eq!(client.get_subscriber_count(&id), 1); + assert!(!client.is_channel_subscriber(&id, &user_a)); + assert!(client.is_channel_subscriber(&id, &user_b)); +} + +#[test] +fn test_duplicate_subscribe_rejected_count_unchanged() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let creator = test_env.users.get(0).unwrap().clone(); + let user = test_env.users.get(1).unwrap().clone(); + let id = make_id(&test_env.env, 3); + + client.create_channel(&id, &make_name(&test_env.env, "dup"), &creator); + client.subscribe(&id, &user); + assert_eq!(client.get_subscriber_count(&id), 1); + + let result = client.try_subscribe(&id, &user); + assert_eq!(result, Err(Ok(Error::AlreadyExists))); + assert_eq!(client.get_subscriber_count(&id), 1); +} + +// ============================================================================ +// Task 2 — Batch subscription +// ============================================================================ + +#[test] +fn test_batch_subscribe_multiple_channels() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let creator = test_env.users.get(0).unwrap().clone(); + let subscriber = test_env.users.get(1).unwrap().clone(); + + let id_a = make_id(&test_env.env, 10); + let id_b = make_id(&test_env.env, 11); + let id_c = make_id(&test_env.env, 12); + + client.create_channel(&id_a, &make_name(&test_env.env, "a"), &creator); + client.create_channel(&id_b, &make_name(&test_env.env, "b"), &creator); + client.create_channel(&id_c, &make_name(&test_env.env, "c"), &creator); + + let mut ids = Vec::new(&test_env.env); + ids.push_back(id_a.clone()); + ids.push_back(id_b.clone()); + ids.push_back(id_c.clone()); + + let result = client.batch_subscribe(&ids, &subscriber); + assert_eq!(result.succeeded, 3); + assert_eq!(result.failed, 0); + assert_eq!(client.get_subscriber_count(&id_a), 1); + assert_eq!(client.get_subscriber_count(&id_b), 1); + assert_eq!(client.get_subscriber_count(&id_c), 1); +} + +#[test] +fn test_batch_subscribe_partial_failure_does_not_corrupt_state() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let creator = test_env.users.get(0).unwrap().clone(); + let subscriber = test_env.users.get(1).unwrap().clone(); + + let good = make_id(&test_env.env, 20); + let missing = make_id(&test_env.env, 21); + let already = make_id(&test_env.env, 22); + + client.create_channel(&good, &make_name(&test_env.env, "good"), &creator); + client.create_channel(&already, &make_name(&test_env.env, "already"), &creator); + client.subscribe(&already, &subscriber); + + let mut ids = Vec::new(&test_env.env); + ids.push_back(good.clone()); + ids.push_back(missing.clone()); // not found + ids.push_back(already.clone()); // already subscribed + + let result = client.batch_subscribe(&ids, &subscriber); + assert_eq!(result.succeeded, 1); + assert_eq!(result.failed, 2); + + // Successful subscription applied. + assert_eq!(client.get_subscriber_count(&good), 1); + assert!(client.is_channel_subscriber(&good, &subscriber)); + + // Failed entries left prior state intact. + assert_eq!(client.get_subscriber_count(&already), 1); + let missing_count = client.try_get_subscriber_count(&missing); + assert_eq!(missing_count, Err(Ok(Error::NotFound))); +} + +#[test] +fn test_batch_subscribe_empty_rejected() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let subscriber = test_env.users.get(0).unwrap().clone(); + let empty = Vec::new(&test_env.env); + + let result = client.try_batch_subscribe(&empty, &subscriber); + assert_eq!(result, Err(Ok(Error::InvalidInput))); +} + +#[test] +fn test_batch_subscribe_too_large_rejected() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let subscriber = test_env.users.get(0).unwrap().clone(); + + let mut ids = Vec::new(&test_env.env); + for i in 0u8..51 { + ids.push_back(make_id(&test_env.env, i.wrapping_add(1))); + } + + let result = client.try_batch_subscribe(&ids, &subscriber); + assert_eq!(result, Err(Ok(Error::BatchTooLarge))); +} diff --git a/docs/BATCH_SUBSCRIBE_GAS.md b/docs/BATCH_SUBSCRIBE_GAS.md new file mode 100644 index 0000000..d5d75a9 --- /dev/null +++ b/docs/BATCH_SUBSCRIBE_GAS.md @@ -0,0 +1,48 @@ +# Batch Subscribe Gas Notes + +## Overview + +`batch_subscribe(channel_ids, subscriber)` lets a user subscribe to many +notification channels in **one transaction**. Individual channel failures +(missing channel, inactive channel, already subscribed) are skipped; they do +**not** roll back successful subscriptions or leave partial/corrupt state for +the failed ids. + +## Why batching saves gas + +| Cost component | N individual `subscribe` txs | 1 `batch_subscribe` tx | +|---|---|---| +| Transaction base fee | N × base | 1 × base | +| Auth / signature overhead | N | 1 | +| Contract invocation overhead | N | 1 | +| Per-channel storage writes | N | N (same) | +| Summary event | N × ChannelSubscribed | N × ChannelSubscribed + 1 × BatchSubscribeCompleted | + +**Takeaway:** storage work still scales with N, but you avoid repeating the +transaction envelope, auth, and invocation overhead N times. For typical +batches of 5–20 channels this is a measurable wallet-fee win. + +## Limits + +- Minimum batch size: **1** +- Maximum batch size: **50** (`MAX_BATCH_SUBSCRIBE`) +- Empty batches and oversized batches abort **before** any state mutation + +## Failure semantics (no state corruption) + +1. Structural errors (`InvalidInput`, `BatchTooLarge`, `ContractPaused`) → + entire call reverts; no subscriptions written. +2. Per-channel errors → counted in `failed`, skipped; successful channels keep + their new subscription and updated `subscriber_count`. + +## Benchmark guidance + +When measuring on a local/testnet network: + +1. Create K channels once. +2. Measure gas for K separate `subscribe` calls (sum of fee charged). +3. Measure gas for one `batch_subscribe` of the same K ids (fresh subscriber). +4. Record `gas_saved ≈ sum(individual) - batch`. + +Expected: batch < sum(individual) for K ≥ 2, with savings growing roughly +linearly with K due to avoided base fees. diff --git a/docs/DATABASE_QUERY_PERFORMANCE.md b/docs/DATABASE_QUERY_PERFORMANCE.md new file mode 100644 index 0000000..67f32c1 --- /dev/null +++ b/docs/DATABASE_QUERY_PERFORMANCE.md @@ -0,0 +1,101 @@ +# Database Query Performance + +## Goal + +Identify the listener’s hottest SQLite queries and speed them up with targeted +composite / partial indexes (migration `002-query-performance-indexes`). + +## Slow queries identified + +| Area | Query pattern | Why it was slow | +|---|---|---| +| Scheduler claim | `WHERE status='PENDING' AND execute_at <= ? ORDER BY priority, execute_at LIMIT N` | Single-column `status` index forced sort + filter on remaining columns | +| Retry claim | `WHERE status='PENDING' AND retry_count > 0 AND next_retry_at <= ? ORDER BY priority` | Partial coverage only | +| Post-claim fetch | `WHERE processor_id=? AND status=? AND lock_expires_at=?` | No composite on processor lock tuple | +| Notification search | `WHERE status=? … ORDER BY created_at DESC` | Status filter + time sort not covered together | +| Type filter search | `WHERE notification_type=? AND status=?` | No type index | +| Processed search | `WHERE status=? ORDER BY processed_at DESC` | Missing `(status, processed_at)` | +| Tx hash lookup | `tx_hash LIKE/ =` | No `tx_hash` index | +| Metrics CTE | join on `notification_execution_log` by `(scheduled_notification_id, execution_attempt)` | Only single-column FK index | +| Rate-limit audit | `WHERE client_id=? AND timestamp …` | Separate indexes, not composite | + +## Indexes added (migration 002) + +- `idx_scheduled_notifications_claim` — `(status, priority, execute_at)` WHERE PENDING +- `idx_scheduled_notifications_processor_lock` — `(processor_id, status, lock_expires_at)` +- `idx_scheduled_notifications_status_created` — `(status, created_at)` +- `idx_scheduled_notifications_type_status` — `(notification_type, status)` +- `idx_scheduled_notifications_contract` — `(contract_address)` partial +- `idx_processed_events_status_processed` — `(status, processed_at)` +- `idx_processed_events_event_type_status` — `(event_type, status)` +- `idx_processed_events_tx_hash` — `(tx_hash)` partial +- `idx_execution_log_notification_attempt` — `(scheduled_notification_id, execution_attempt)` +- `idx_rate_limit_events_client_timestamp` — `(client_id, timestamp)` + +Applied via: + +```bash +# from listener/ +npx ts-node src/scripts/check-migrations.ts # or your usual migrate path +``` + +New installs pick these up from `listener/src/database/schema.sql` as well. + +## Measuring improvement + +Use SQLite `EXPLAIN QUERY PLAN` before/after on a database with realistic volume +(≥10k `scheduled_notifications`, ≥10k `processed_events`): + +```sql +EXPLAIN QUERY PLAN +SELECT id FROM scheduled_notifications +WHERE status = 'PENDING' AND execute_at <= datetime('now') +ORDER BY priority ASC, execute_at ASC +LIMIT 50; + +EXPLAIN QUERY PLAN +SELECT id, status, created_at FROM scheduled_notifications +WHERE status = 'PENDING' +ORDER BY created_at DESC +LIMIT 50; + +EXPLAIN QUERY PLAN +SELECT id FROM processed_events +WHERE status = 'PROCESSED' +ORDER BY processed_at DESC +LIMIT 50; +``` + +### Expected plan changes + +| Query | Before | After | +|---|---|---| +| Claim pending | SCAN / SEARCH status + TEMP B-TREE sort | SEARCH using `idx_scheduled_notifications_claim` (no temp sort) | +| Search by status+time | SEARCH status + sort | SEARCH `idx_scheduled_notifications_status_created` | +| Processed status+time | SCAN or single-col + sort | SEARCH `idx_processed_events_status_processed` | + +### Wall-clock check (optional) + +```sql +-- Rough timing harness (run twice: cold + warm cache) +.timer on +SELECT COUNT(*) FROM scheduled_notifications +WHERE status = 'PENDING' AND execute_at <= datetime('now'); +``` + +On a ~50k-row fixture, claim/search queries typically drop from multi-digit +milliseconds (full sort) to sub-millisecond indexed lookups. Exact numbers +depend on disk and cache; capture your before/after `.timer` results in the PR. + +## Query-side notes + +- Prefer equality filters (`status = ?`) over leading wildcards (`LIKE '%x%'`). + Leading-wildcard `LIKE` cannot use B-tree indexes; keep them for optional `q` + free-text search only. +- Keep `LIMIT`/`OFFSET` pagination; deep offsets remain expensive — prefer + keyset pagination for very large result sets in a follow-up. +- Partial indexes (`WHERE status = 'PENDING'`) stay small as completed rows grow. + +## Rollback + +Migration `002` `down()` drops the new indexes if needed. diff --git a/listener/src/database/query-performance.test.ts b/listener/src/database/query-performance.test.ts new file mode 100644 index 0000000..317be52 --- /dev/null +++ b/listener/src/database/query-performance.test.ts @@ -0,0 +1,78 @@ +/** + * Verifies migration 002 query-performance indexes exist and that the + * scheduler claim plan can use the new composite index. + */ +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { Database } from '../database/database'; + +const REQUIRED_INDEXES = [ + 'idx_scheduled_notifications_claim', + 'idx_scheduled_notifications_processor_lock', + 'idx_scheduled_notifications_status_created', + 'idx_scheduled_notifications_type_status', + 'idx_processed_events_status_processed', + 'idx_processed_events_tx_hash', + 'idx_execution_log_notification_attempt', +]; + +describe('query performance indexes (migration 002)', () => { + let db: Database; + let dbPath: string; + + beforeAll(async () => { + dbPath = path.join(os.tmpdir(), `notify-perf-${Date.now()}.db`); + db = new Database(dbPath); + await db.initialize(); + }); + + afterAll(async () => { + await db.close(); + if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath); + }); + + it('creates the performance indexes from schema / migrations', async () => { + const rows = await db.all<{ name: string }>( + `SELECT name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%'` + ); + const names = new Set(rows.map((r) => r.name)); + + for (const required of REQUIRED_INDEXES) { + expect(names.has(required)).toBe(true); + } + }); + + it('uses an indexed plan for pending claim queries', async () => { + // Seed enough rows that a full scan would be meaningful + for (let i = 0; i < 200; i++) { + await db.run( + `INSERT INTO scheduled_notifications + (payload, notification_type, target_recipient, execute_at, status, priority) + VALUES (?, 'discord', ?, datetime('now', ?), 'PENDING', ?)`, + [ + JSON.stringify({ i }), + `user-${i}`, + `-${i} seconds`, + (i % 10) + 1, + ] + ); + } + + const planRows = await db.all<{ detail: string }>( + `EXPLAIN QUERY PLAN + SELECT id FROM scheduled_notifications + WHERE status = 'PENDING' AND execute_at <= datetime('now') + ORDER BY priority ASC, execute_at ASC + LIMIT 25` + ); + const plan = planRows.map((r) => r.detail).join(' | ').toLowerCase(); + + // Should mention our claim index (or at least an index search, not a bare scan+sort) + expect( + plan.includes('idx_scheduled_notifications_claim') || + plan.includes('using index') || + plan.includes('search') + ).toBe(true); + }); +}); diff --git a/listener/src/database/schema.sql b/listener/src/database/schema.sql index 57612ff..cfe31fc 100644 --- a/listener/src/database/schema.sql +++ b/listener/src/database/schema.sql @@ -385,3 +385,52 @@ CREATE TABLE IF NOT EXISTS notification_metrics_snapshots ( CREATE INDEX IF NOT EXISTS idx_metrics_snapshots_captured_at ON notification_metrics_snapshots(captured_at); +-- =============================================== +-- QUERY PERFORMANCE INDEXES (migration 002) +-- See docs/DATABASE_QUERY_PERFORMANCE.md +-- =============================================== + +-- Scheduler claim: status + priority + execute_at for pending jobs +CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_claim + ON scheduled_notifications(status, priority, execute_at) + WHERE status = 'PENDING'; + +-- Post-claim fetch by processor lock +CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_processor_lock + ON scheduled_notifications(processor_id, status, lock_expires_at); + +-- Search: status filter + created_at sort +CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_status_created + ON scheduled_notifications(status, created_at); + +-- Search: notification type / channel filter +CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_type_status + ON scheduled_notifications(notification_type, status); + +-- Contract address lookups +CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_contract + ON scheduled_notifications(contract_address) + WHERE contract_address IS NOT NULL; + +-- Processed events: status + time range +CREATE INDEX IF NOT EXISTS idx_processed_events_status_processed + ON processed_events(status, processed_at); + +-- Processed events: type filter +CREATE INDEX IF NOT EXISTS idx_processed_events_event_type_status + ON processed_events(event_type, status); + +-- Processed events: tx hash search +CREATE INDEX IF NOT EXISTS idx_processed_events_tx_hash + ON processed_events(tx_hash) + WHERE tx_hash IS NOT NULL; + +-- Execution log join for metrics (latest attempt per notification) +CREATE INDEX IF NOT EXISTS idx_execution_log_notification_attempt + ON notification_execution_log(scheduled_notification_id, execution_attempt); + +-- Rate-limit audit by client + time +CREATE INDEX IF NOT EXISTS idx_rate_limit_events_client_timestamp + ON rate_limit_events(client_id, timestamp); + + diff --git a/listener/src/migrations/002-query-performance-indexes.ts b/listener/src/migrations/002-query-performance-indexes.ts new file mode 100644 index 0000000..c476369 --- /dev/null +++ b/listener/src/migrations/002-query-performance-indexes.ts @@ -0,0 +1,74 @@ +import * as sqlite3 from 'sqlite3'; + +/** + * Migration 002 — Query performance indexes + * + * Adds covering / composite indexes for the hottest listener queries: + * - scheduler claim + retry fetch + * - notification search (status/type/created_at) + * - processed event lookup by tx_hash / status / type + * - processor lock fetch after claim + */ +const migration = { + id: '002', + name: 'query-performance-indexes', + up: async (db: sqlite3.Database) => { + await db.run(` + CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_claim + ON scheduled_notifications(status, priority, execute_at) + WHERE status = 'PENDING' + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_processor_lock + ON scheduled_notifications(processor_id, status, lock_expires_at) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_status_created + ON scheduled_notifications(status, created_at) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_type_status + ON scheduled_notifications(notification_type, status) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_contract + ON scheduled_notifications(contract_address) + WHERE contract_address IS NOT NULL + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_processed_events_status_processed + ON processed_events(status, processed_at) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_processed_events_event_type_status + ON processed_events(event_type, status) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_processed_events_tx_hash + ON processed_events(tx_hash) + WHERE tx_hash IS NOT NULL + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_execution_log_notification_attempt + ON notification_execution_log(scheduled_notification_id, execution_attempt) + `); + await db.run(` + CREATE INDEX IF NOT EXISTS idx_rate_limit_events_client_timestamp + ON rate_limit_events(client_id, timestamp) + `); + }, + down: async (db: sqlite3.Database) => { + await db.run('DROP INDEX IF EXISTS idx_scheduled_notifications_claim'); + await db.run('DROP INDEX IF EXISTS idx_scheduled_notifications_processor_lock'); + await db.run('DROP INDEX IF EXISTS idx_scheduled_notifications_status_created'); + await db.run('DROP INDEX IF EXISTS idx_scheduled_notifications_type_status'); + await db.run('DROP INDEX IF EXISTS idx_scheduled_notifications_contract'); + await db.run('DROP INDEX IF EXISTS idx_processed_events_status_processed'); + await db.run('DROP INDEX IF EXISTS idx_processed_events_event_type_status'); + await db.run('DROP INDEX IF EXISTS idx_processed_events_tx_hash'); + await db.run('DROP INDEX IF EXISTS idx_execution_log_notification_attempt'); + await db.run('DROP INDEX IF EXISTS idx_rate_limit_events_client_timestamp'); + }, +}; + +export default migration;