Skip to content
HeliosDB HeliosDB Nano

Real-Time Pub/Sub (LISTEN/NOTIFY): Business Use Case for HeliosDB Nano

Document ID: 12_REALTIME_PUBSUB.md Created: 2025-11-30 Category: Event-Driven Architecture

Executive Summary

PostgreSQL-compatible LISTEN/NOTIFY pub/sub is planned for HeliosDB Nano; it is not in the current release. The planned design targets in-process event delivery across many concurrent channels without external dependencies. With thread-safe in-memory channel management, automatic subscription cleanup, and 8KB payload support, the planned design would let event-driven applications implement real-time dashboards, cache invalidation, activity feeds, workflow automation, and microservices coordination entirely within embedded, edge, and microservice deployments without Redis, Kafka, or message queue infrastructure. The planned zero-external-dependency design would avoid network hops for local notifications and the cost of running a separate broker, and would enable offline-first event-driven applications for IoT devices, collaborative tools, and real-time analytics platforms.

Problem Being Solved

Core Problem Statement

Modern applications require real-time event notification for responsive user interfaces, cache coherence, workflow orchestration, and microservices coordination, but existing solutions force teams to choose between heavyweight message brokers (Redis Pub/Sub, Kafka, RabbitMQ) requiring separate infrastructure and network overhead, or inefficient polling-based architectures that waste CPU cycles and increase latency. Teams deploying to edge devices, embedded systems, or resource-constrained environments cannot afford external message queue dependencies, yet lack lightweight pub/sub for event-driven patterns.

Root Cause Analysis

FactorImpactCurrent WorkaroundLimitation
External Message Broker Dependency5-50ms network latency per message, $100-1000/month infrastructure cost, complex deploymentDeploy Redis Pub/Sub, Kafka, or RabbitMQ as separate serviceRequires network connectivity, adds operational complexity, unsuitable for embedded/edge deployments, single point of failure
Database Polling Anti-PatternHigh CPU usage (10-30% wasted on polling), 100-1000ms notification delay, increased database loadPoll database every 100ms-1s with SELECT * FROM events WHERE timestamp > last_checkWastes resources, poor scalability, delayed notifications, database contention, 10-100x slower than push-based
PostgreSQL Server Requirement200MB+ memory overhead, complex deployment, requires client-server architectureDeploy full PostgreSQL server for LISTEN/NOTIFY500MB+ footprint, unsuitable for embedded systems, client-server latency penalty, complex connection management
WebSocket/Long-Polling OverheadRequires separate HTTP server, complex state management, high memory per connectionImplement custom WebSocket server or long-polling endpoints1-5MB memory per connection, stateful server complexity, doesn’t integrate with database transactions
In-Memory Event Bus LimitationsNo persistence, no integration with database transactions, manual delivery guaranteesUse in-app event emitters (EventEmitter, Tokio broadcast channels)No transaction integration, lost messages on crash, no database trigger support, manual coordination

Business Impact (design targets)

The “With HeliosDB Nano” column states design targets for the planned feature; none of these figures has been measured.

MetricWithout HeliosDB NanoWith HeliosDB Nano (target)Improvement (target)
Notification Latency (in-process)5-50ms (external broker) + network<1ms (in-memory)5-50x faster
Infrastructure Cost$100-1000/month (Redis/Kafka)$0 (embedded)100% reduction
Memory Footprint (1000 channels)200MB+ (Redis) or 500MB+ (Postgres)<10MB (in-memory)20-50x reduction
Deployment Complexity3-5 services (message broker, DB, app)Single binary70% simpler
CPU Overhead (polling replaced)10-30% wasted on polling<1% (event-driven)10-30x more efficient
Edge Device ViabilityImpossible (requires external broker)Full support (Raspberry Pi+)Enables new markets

Who Suffers Most

  1. Real-Time Dashboard Teams: Building analytics dashboards that update live on data changes, forced to poll databases every second, wasting 20-40% CPU and experiencing 500ms-2s update delays when sub-second responsiveness is critical for user experience.

  2. Cache Invalidation Engineers: Managing cache coherence across distributed application instances without Redis Pub/Sub infrastructure, resulting in stale cache issues, manual cache TTL management, or expensive cache-aside patterns with 10-100x more cache misses than necessary.

  3. IoT/Edge Application Developers: Building collaborative tools or real-time monitoring on edge devices where external message brokers are unavailable, forcing them to implement inefficient polling or abandon real-time features entirely, degrading user experience significantly.

Why Competitors Cannot Solve This

Technical Barriers

Competitor CategoryLimitationRoot CauseTime to Match
SQLite, DuckDBNo pub/sub or notification mechanismDesigned for single-process analytical workloads; no multi-client notification infrastructure12-18 months
Redis Pub/SubRequires separate server, network overhead, no database transaction integrationCloud/server-first architecture designed for distributed systems, not embedded use casesNever (contradicts embedded model)
PostgreSQL LISTEN/NOTIFYRequires full Postgres server (500MB+ footprint), client-server architectureFull RDBMS designed for client-server deployment, not in-process embedding6-12 months for embedded variant
Kafka, RabbitMQHeavy infrastructure (1GB+ memory), complex deployment, designed for durability not speedEnterprise message broker architecture prioritizing durability over embedded simplicityNever (enterprise-focused)
WebSocket ServersSeparate HTTP infrastructure, no database integration, stateful connection managementApplication-layer protocol disconnected from database layer18-24 months to add DB integration

Architecture Requirements

To match HeliosDB Nano’s real-time pub/sub capabilities, competitors would need:

  1. In-Process Thread-Safe Channel Management: Build concurrent hash map for channel subscriptions with read-write locks, UUID-based subscription tracking, and automatic cleanup on connection drop, requiring deep understanding of Rust concurrency primitives (Arc, RwLock) and lock-free data structures to avoid deadlocks while supporting thousands of channels.

  2. Integration with Database Transaction Lifecycle: Implement NOTIFY delivery that respects transaction boundaries (only deliver after COMMIT, not on ROLLBACK), coordinate with SQL executor to trigger notifications on database mutations, and handle trigger-based notifications, requiring integration with transaction state machine and SQL parser.

  3. PostgreSQL Wire Protocol Compatibility: Implement asynchronous notification messages in extended query protocol, support LISTEN/UNLISTEN SQL commands, handle notification delivery during query idle periods, and maintain compatibility with PostgreSQL client libraries (psycopg2, pg, node-postgres), requiring protocol-level implementation expertise.

HeliosDB Nano Solution

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│              Application Code (Multi-threaded)               │
├──────────────────┬──────────────────┬───────────────────────┤
│  LISTEN Thread   │  NOTIFY Thread   │  Business Logic Thread │
│  (Subscribe)     │  (Publish)       │  (SQL Queries)         │
├──────────────────┴──────────────────┴───────────────────────┤
│              PubSubManager (Thread-Safe)                     │
│  ┌─────────────────────────────────────────────────────┐    │
│  │  Channel → Subscribers (HashMap<String, HashSet>)   │    │
│  │  Subscription → Notification Queue (HashMap<Uuid>)  │    │
│  │  Arc<RwLock> for concurrent read/write              │    │
│  └─────────────────────────────────────────────────────┘    │
├──────────────────────────────────────────────────────────────┤
│        SQL Executor (Transaction Integration)                │
│  - LISTEN/UNLISTEN command parsing                           │
│  - NOTIFY command execution                                  │
│  - Database trigger → NOTIFY integration                     │
├──────────────────────────────────────────────────────────────┤
│           HeliosDB Nano Storage Engine                       │
└──────────────────────────────────────────────────────────────┘


In-Memory Notification Flow:
  1. Thread A: LISTEN 'orders'  →  Subscribe to 'orders' channel
  2. Thread B: NOTIFY 'orders', 'payload'  →  Enqueue notification
  3. Thread A: Poll subscription  →  Receive notification (<1ms)

Key Capabilities

CapabilityDescriptionDesign target (not measured)
PostgreSQL-Compatible LISTEN/NOTIFYSQL commands for subscribing (LISTEN channel) and publishing (NOTIFY channel, 'payload') with identical syntax to PostgreSQL100% PostgreSQL syntax compatibility
In-Memory Channel ManagementThread-safe concurrent HashMap for channels and subscriptions using Arc for lock-free readsThousands of concurrent channels, <10MB overhead
Sub-Millisecond Notification DeliveryIn-process notification queue delivery without network overhead or serialization<1ms P99 latency for in-process delivery
Automatic Subscription CleanupRAII-based subscription handles with Drop trait implementation for automatic UNLISTENZero memory leaks, automatic resource management
8KB Payload SupportNotification payloads up to 8000 bytes (PostgreSQL limit) for rich event data (JSON, binary)Full PostgreSQL payload compatibility
Multiple Subscribers per ChannelBroadcast notifications to all subscribers on a channel (fan-out pattern)O(N) delivery to N subscribers
Transaction IntegrationNOTIFY respects transaction boundaries (only delivered after COMMIT, not on ROLLBACK)ACID-compliant event delivery
Database Trigger SupportTrigger functions can call NOTIFY for automatic event generation on INSERT/UPDATE/DELETEZero-code change detection

Concrete Examples with Code, Config & Architecture

Example 1: Real-Time Dashboard Updates - Embedded Configuration

Scenario: Analytics dashboard displaying live sales metrics for 1000 concurrent users, requiring sub-second updates when new orders arrive without polling database every 100ms (which wastes 20% CPU).

Architecture:

Web Browser (WebSocket)
    ↓
Web Server (Axum/Actix)
    ↓
HeliosDB Nano (LISTEN 'sales_updates')
    ↓
In-Process Notification Queue
    ↓
Database Trigger on INSERT → NOTIFY 'sales_updates'

Configuration (heliosdb.toml):

# HeliosDB Nano configuration for real-time dashboard
[database]
path = "/var/lib/heliosdb/sales.db"
memory_limit_mb = 512
enable_wal = true


[pubsub]
# Enable pub/sub with optimized settings
enabled = true
max_channels = 1000
max_payload_bytes = 8000
notification_queue_size = 100


[monitoring]
metrics_enabled = true
verbose_logging = false

Implementation Code (Rust with Axum WebSocket):

use heliosdb_nano::{Connection, protocols::PubSubManager};
use axum::{
    extract::{ws::WebSocket, State, WebSocketUpgrade},
    response::Response,
    routing::get,
    Router,
};
use std::sync::Arc;
use tokio::sync::broadcast;


#[derive(Clone)]
struct AppState {
    db: Arc<Connection>,
    pubsub: Arc<PubSubManager>,
    broadcast_tx: broadcast::Sender<String>,
}


// Initialize database with trigger for automatic notifications
fn init_dashboard_db(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
    // Create sales table
    conn.execute(
        "CREATE TABLE IF NOT EXISTS sales_orders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id INTEGER NOT NULL,
            amount DECIMAL(10,2) NOT NULL,
            created_at INTEGER DEFAULT (strftime('%s', 'now'))
        )",
        [],
    )?;


    // Create trigger to automatically NOTIFY on new sales
    conn.execute(
        "CREATE TRIGGER IF NOT EXISTS sales_order_notify
         AFTER INSERT ON sales_orders
         BEGIN
             SELECT NOTIFY('sales_updates',
                 json_object(
                     'order_id', NEW.id,
                     'customer_id', NEW.customer_id,
                     'amount', NEW.amount,
                     'timestamp', NEW.created_at
                 )
             );
         END",
        [],
    )?;


    Ok(())
}


// Background task that listens for database notifications and broadcasts to WebSocket clients
async fn notification_broadcaster(
    pubsub: Arc<PubSubManager>,
    tx: broadcast::Sender<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    // Subscribe to sales_updates channel
    let subscription = pubsub.subscribe("sales_updates")?;


    loop {
        // Poll for new notifications (non-blocking)
        let notifications = subscription.poll()?;


        for notification in notifications {
            // Broadcast to all WebSocket clients
            let _ = tx.send(notification.payload);
        }


        // Small delay to batch notifications
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    }
}


// WebSocket handler for individual client connections
async fn websocket_handler(
    ws: WebSocketUpgrade,
    State(state): State<AppState>,
) -> Response {
    ws.on_upgrade(|socket| handle_websocket(socket, state))
}


async fn handle_websocket(mut socket: WebSocket, state: AppState) {
    // Subscribe to broadcast channel
    let mut rx = state.broadcast_tx.subscribe();


    // Forward notifications to WebSocket client
    while let Ok(message) = rx.recv().await {
        if socket.send(axum::extract::ws::Message::Text(message)).await.is_err() {
            break;
        }
    }
}


// API endpoint to insert new sales orders (triggers notification)
async fn create_order(
    State(state): State<AppState>,
    axum::Json(order): axum::Json<serde_json::Value>,
) -> Result<String, String> {
    state.db.execute(
        "INSERT INTO sales_orders (customer_id, amount) VALUES (?1, ?2)",
        [
            order["customer_id"].as_i64().unwrap().to_string(),
            order["amount"].as_f64().unwrap().to_string(),
        ],
    ).map_err(|e| e.to_string())?;


    Ok("Order created".to_string())
}


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize database
    let conn = Arc::new(Connection::open("sales.db")?);
    init_dashboard_db(&conn)?;


    // Initialize pub/sub manager
    let pubsub = Arc::new(PubSubManager::new());


    // Create broadcast channel for WebSocket fanout
    let (broadcast_tx, _) = broadcast::channel(1000);


    let state = AppState {
        db: conn,
        pubsub: Arc::clone(&pubsub),
        broadcast_tx: broadcast_tx.clone(),
    };


    // Start notification broadcaster task
    let broadcaster_pubsub = Arc::clone(&pubsub);
    tokio::spawn(async move {
        notification_broadcaster(broadcaster_pubsub, broadcast_tx).await
    });


    // Build web server
    let app = Router::new()
        .route("/ws", get(websocket_handler))
        .route("/orders", axum::routing::post(create_order))
        .with_state(state);


    // Start server
    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    println!("Dashboard server running on http://0.0.0.0:8080");
    axum::serve(listener, app).await?;


    Ok(())
}

Design targets (not measured; feature planned):

MetricBefore (Polling)After (LISTEN/NOTIFY)Improvement
Update Latency500-1000ms (polling interval)<50ms (event-driven)10-20x faster
CPU Usage (idle)20% (constant polling)<1% (event-driven)20x more efficient
Database Load10 queries/sec per client0 queries/sec (notifications only)100% reduction
Scalability100 concurrent clients max1000+ concurrent clients10x improvement
Infrastructure Cost$200/month (Redis + DB)$50/month (DB only)75% reduction

Example 2: Cache Invalidation - Python Integration

Scenario: Multi-instance web application with in-memory caches that must invalidate when source data changes, without polling or manual TTL management.

Python Client Code:

import heliosdb_nano
from heliosdb_nano import Connection, PubSubManager
import asyncio
import json
from typing import Dict, Any
import threading


class CacheInvalidationManager:
    """Manages cache with automatic invalidation via LISTEN/NOTIFY."""


    def __init__(self, db_path: str):
        self.conn = Connection.open(db_path)
        self.pubsub = PubSubManager()
        self.cache: Dict[str, Any] = {}
        self.lock = threading.Lock()


    def setup_schema(self):
        """Initialize database schema with invalidation triggers."""
        # Create products table
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS products (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                price DECIMAL(10,2) NOT NULL,
                updated_at INTEGER DEFAULT (strftime('%s', 'now'))
            )
        """)


        # Create trigger to notify on UPDATE
        self.conn.execute("""
            CREATE TRIGGER IF NOT EXISTS product_update_notify
            AFTER UPDATE ON products
            BEGIN
                SELECT NOTIFY('cache_invalidate',
                    'product:' || NEW.id
                );
            END
        """)


        # Create trigger to notify on DELETE
        self.conn.execute("""
            CREATE TRIGGER IF NOT EXISTS product_delete_notify
            AFTER DELETE ON products
            BEGIN
                SELECT NOTIFY('cache_invalidate',
                    'product:' || OLD.id
                );
            END
        """)


    async def start_invalidation_listener(self):
        """Background task that listens for cache invalidation events."""
        # Subscribe to cache invalidation channel
        subscription = self.pubsub.subscribe('cache_invalidate')


        while True:
            # Poll for notifications
            notifications = subscription.poll()


            for notification in notifications:
                cache_key = notification.payload


                # Invalidate cache entry
                with self.lock:
                    if cache_key in self.cache:
                        del self.cache[cache_key]
                        print(f"Cache invalidated: {cache_key}")


            # Poll every 10ms
            await asyncio.sleep(0.01)


    def get_product(self, product_id: int) -> Dict[str, Any]:
        """Get product with caching."""
        cache_key = f"product:{product_id}"


        # Check cache first
        with self.lock:
            if cache_key in self.cache:
                print(f"Cache HIT: {cache_key}")
                return self.cache[cache_key]


        # Cache miss - query database
        print(f"Cache MISS: {cache_key}")
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT id, name, price FROM products WHERE id = ?",
            (product_id,)
        )


        row = cursor.fetchone()
        if row:
            product = {
                'id': row[0],
                'name': row[1],
                'price': row[2]
            }


            # Store in cache
            with self.lock:
                self.cache[cache_key] = product


            return product


        return None


    def update_product(self, product_id: int, name: str, price: float):
        """Update product (triggers cache invalidation via NOTIFY)."""
        self.conn.execute(
            "UPDATE products SET name = ?, price = ?, updated_at = strftime('%s', 'now') WHERE id = ?",
            (name, price, product_id)
        )
        # Trigger fires automatically, sending NOTIFY


# Usage example
async def main():
    manager = CacheInvalidationManager("products.db")
    manager.setup_schema()


    # Insert test data
    manager.conn.execute(
        "INSERT OR REPLACE INTO products (id, name, price) VALUES (1, 'Widget', 9.99)"
    )


    # Start background invalidation listener
    asyncio.create_task(manager.start_invalidation_listener())


    # First access - cache miss
    product = manager.get_product(1)  # Cache MISS
    print(f"Product: {product}")


    # Second access - cache hit
    product = manager.get_product(1)  # Cache HIT
    print(f"Product: {product}")


    # Update product (triggers invalidation)
    manager.update_product(1, 'Super Widget', 19.99)


    # Wait for invalidation to propagate
    await asyncio.sleep(0.1)


    # Third access - cache miss after invalidation
    product = manager.get_product(1)  # Cache MISS (invalidated)
    print(f"Product: {product}")


    # Keep running
    await asyncio.sleep(3600)


if __name__ == "__main__":
    asyncio.run(main())

Architecture Pattern:

┌─────────────────────────────────────────┐
│  Application Instance 1                  │
│  ┌────────────────┐  ┌────────────────┐ │
│  │ In-Memory Cache│  │ LISTEN Thread  │ │
│  └────────────────┘  └────────────────┘ │
├─────────────────────────────────────────┤
│  Application Instance 2                  │
│  ┌────────────────┐  ┌────────────────┐ │
│  │ In-Memory Cache│  │ LISTEN Thread  │ │
│  └────────────────┘  └────────────────┘ │
├─────────────────────────────────────────┤
│         HeliosDB Nano (Shared)           │
│  Database Trigger → NOTIFY all listeners │
└─────────────────────────────────────────┘


Flow:
1. Instance 1: UPDATE products SET price = 19.99 WHERE id = 1
2. Database Trigger: NOTIFY 'cache_invalidate', 'product:1'
3. All Instances: Receive notification, invalidate local cache
4. Next read: Cache miss, fetch fresh data

Design targets (not measured; feature planned):

  • Cache hit rate: 95% (vs 60% with TTL-based invalidation)
  • Stale cache duration: <50ms (vs 30-60s with TTL)
  • Cache memory efficiency: 80% reduction (no defensive TTL padding)
  • Infrastructure cost: $0 (no Redis required)

Example 3: Activity Feed - Microservices Integration (Go/Rust)

Scenario: Social platform with 10,000 users, displaying real-time activity feed updates (likes, comments, follows) without polling every user’s feed.

Rust Service Code (src/activity_service.rs):

use axum::{
    extract::{Path, State},
    http::StatusCode,
    routing::{get, post},
    Json, Router,
};
use heliosdb_nano::{Connection, protocols::PubSubManager};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::broadcast;


#[derive(Clone)]
pub struct ActivityService {
    db: Arc<Connection>,
    pubsub: Arc<PubSubManager>,
}


#[derive(Debug, Serialize, Deserialize)]
pub struct Activity {
    id: i64,
    user_id: i64,
    action_type: String,  // "like", "comment", "follow"
    target_id: i64,
    timestamp: i64,
}


impl ActivityService {
    pub fn new(db_path: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let db = Arc::new(Connection::open(db_path)?);
        let pubsub = Arc::new(PubSubManager::new());


        // Initialize schema with triggers
        Self::init_schema(&db)?;


        Ok(Self { db, pubsub })
    }


    fn init_schema(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
        // Create activities table
        conn.execute(
            "CREATE TABLE IF NOT EXISTS activities (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                action_type TEXT NOT NULL,
                target_id INTEGER NOT NULL,
                timestamp INTEGER DEFAULT (strftime('%s', 'now'))
            )",
            [],
        )?;


        // Create trigger to notify followers on new activity
        conn.execute(
            "CREATE TRIGGER IF NOT EXISTS activity_notify
             AFTER INSERT ON activities
             BEGIN
                 SELECT NOTIFY(
                     'activity_feed_' || NEW.user_id,
                     json_object(
                         'id', NEW.id,
                         'user_id', NEW.user_id,
                         'action_type', NEW.action_type,
                         'target_id', NEW.target_id,
                         'timestamp', NEW.timestamp
                     )
                 );
             END",
            [],
        )?;


        Ok(())
    }


    // Record a new activity (automatically triggers notification)
    pub fn record_activity(
        &self,
        user_id: i64,
        action_type: &str,
        target_id: i64,
    ) -> Result<Activity, Box<dyn std::error::Error>> {
        let mut stmt = self.db.prepare(
            "INSERT INTO activities (user_id, action_type, target_id)
             VALUES (?1, ?2, ?3)
             RETURNING id, user_id, action_type, target_id, timestamp"
        )?;


        let activity = stmt.query_row(
            [user_id.to_string(), action_type.to_string(), target_id.to_string()],
            |row| {
                Ok(Activity {
                    id: row.get(0)?,
                    user_id: row.get(1)?,
                    action_type: row.get(2)?,
                    target_id: row.get(3)?,
                    timestamp: row.get(4)?,
                })
            },
        )?;


        Ok(activity)
    }


    // Subscribe to a user's activity feed
    pub fn subscribe_to_feed(&self, user_id: i64)
        -> Result<Box<dyn heliosdb_nano::protocols::SubscriptionHandle>, Box<dyn std::error::Error>>
    {
        let channel = format!("activity_feed_{}", user_id);
        let subscription = self.pubsub.subscribe(&channel)?;
        Ok(subscription)
    }
}


// HTTP handlers
async fn post_activity(
    State(service): State<Arc<ActivityService>>,
    Json(req): Json<serde_json::Value>,
) -> (StatusCode, Json<Activity>) {
    let user_id = req["user_id"].as_i64().unwrap();
    let action_type = req["action_type"].as_str().unwrap();
    let target_id = req["target_id"].as_i64().unwrap();


    let activity = service.record_activity(user_id, action_type, target_id).unwrap();


    (StatusCode::CREATED, Json(activity))
}


async fn get_activity_stream(
    Path(user_id): Path<i64>,
    State(service): State<Arc<ActivityService>>,
) -> Result<axum::response::sse::Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, std::convert::Infallible>>>, StatusCode> {
    use futures::stream::StreamExt;


    // Subscribe to user's activity feed
    let subscription = service.subscribe_to_feed(user_id)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;


    // Create SSE stream
    let stream = futures::stream::unfold(subscription, |sub| async move {
        loop {
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;


            if let Ok(notifications) = sub.poll() {
                for notification in notifications {
                    let event = axum::response::sse::Event::default()
                        .data(notification.payload);
                    return Some((Ok(event), sub));
                }
            }
        }
    });


    Ok(axum::response::sse::Sse::new(stream))
}


pub fn create_router(service: Arc<ActivityService>) -> Router {
    Router::new()
        .route("/activities", post(post_activity))
        .route("/activities/stream/:user_id", get(get_activity_stream))
        .with_state(service)
}


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let service = Arc::new(ActivityService::new("activities.db")?);
    let app = create_router(service);


    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    println!("Activity service running on http://0.0.0.0:8080");
    axum::serve(listener, app).await?;


    Ok(())
}

Service Architecture:

┌─────────────────────────────────────────┐
│     Client (Browser / Mobile App)       │
│     Server-Sent Events (SSE) Stream     │
├─────────────────────────────────────────┤
│     Activity Microservice (Rust)        │
│  ┌────────────────────────────────────┐ │
│  │  POST /activities                  │ │
│  │    ↓                               │ │
│  │  INSERT INTO activities            │ │
│  │    ↓                               │ │
│  │  Trigger fires NOTIFY              │ │
│  └────────────────────────────────────┘ │
│  ┌────────────────────────────────────┐ │
│  │  GET /activities/stream/:user_id   │ │
│  │    ↓                               │ │
│  │  LISTEN 'activity_feed_123'        │ │
│  │    ↓                               │ │
│  │  Poll subscription → SSE stream    │ │
│  └────────────────────────────────────┘ │
├─────────────────────────────────────────┤
│         HeliosDB Nano (In-Process)      │
│  PubSubManager → Notification Delivery  │
└─────────────────────────────────────────┘

Design targets (not measured; feature planned):

  • Real-time update latency: <100ms (vs 1-5s with polling)
  • Memory per connection: 10KB (vs 1MB with WebSocket state)
  • CPU usage: <5% for 1000 concurrent streams
  • Infrastructure: Single service (no separate message broker)

Example 4: Workflow Automation - Docker/Container Deployment

Scenario: Order processing workflow where each step triggers the next automatically via database events, eliminating manual coordination.

Docker Deployment (Dockerfile):

FROM rust:latest as builder


WORKDIR /app
COPY . .
RUN cargo build --release


FROM debian:bookworm-slim


RUN apt-get update && apt-get install -y \
    ca-certificates \
    && rm -rf /var/lib/apt/lists/*


COPY --from=builder /app/target/release/workflow-engine /usr/local/bin/


RUN mkdir -p /data


EXPOSE 8080


VOLUME ["/data"]


ENTRYPOINT ["workflow-engine"]
CMD ["--config", "/etc/workflow/config.toml"]

Docker Compose (docker-compose.yml):

version: '3.8'


services:
  workflow-engine:
    build:
      context: .
      dockerfile: Dockerfile
    image: workflow-engine:latest


    ports:
      - "8080:8080"


    volumes:
      - ./data:/data
      - ./config/workflow.toml:/etc/workflow/config.toml:ro


    environment:
      RUST_LOG: "workflow_engine=info"
      HELIOSDB_DATA_DIR: "/data"


    restart: unless-stopped


    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 3s
      retries: 3


networks:
  default:
    driver: bridge

Workflow Configuration (config/workflow.toml):

[database]
path = "/data/workflow.db"
memory_limit_mb = 256
enable_wal = true


[pubsub]
enabled = true
max_channels = 100


[workflow]
# Workflow step definitions
[workflow.steps.payment_processing]
trigger_channel = "order_created"
action = "process_payment"
next_channel = "payment_completed"


[workflow.steps.inventory_allocation]
trigger_channel = "payment_completed"
action = "allocate_inventory"
next_channel = "inventory_allocated"


[workflow.steps.shipping]
trigger_channel = "inventory_allocated"
action = "create_shipment"
next_channel = "shipment_created"

Workflow Engine Code (src/workflow.rs):

use heliosdb_nano::{Connection, protocols::PubSubManager};
use std::sync::Arc;
use tokio::task::JoinSet;


pub struct WorkflowEngine {
    db: Arc<Connection>,
    pubsub: Arc<PubSubManager>,
}


impl WorkflowEngine {
    pub fn new(db_path: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let db = Arc::new(Connection::open(db_path)?);
        let pubsub = Arc::new(PubSubManager::new());


        Self::init_schema(&db)?;


        Ok(Self { db, pubsub })
    }


    fn init_schema(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
        // Orders table
        conn.execute(
            "CREATE TABLE IF NOT EXISTS orders (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                customer_id INTEGER NOT NULL,
                amount DECIMAL(10,2) NOT NULL,
                status TEXT DEFAULT 'pending',
                created_at INTEGER DEFAULT (strftime('%s', 'now'))
            )",
            [],
        )?;


        // Workflow events table
        conn.execute(
            "CREATE TABLE IF NOT EXISTS workflow_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                order_id INTEGER NOT NULL,
                event_type TEXT NOT NULL,
                payload TEXT,
                created_at INTEGER DEFAULT (strftime('%s', 'now'))
            )",
            [],
        )?;


        // Trigger on order creation
        conn.execute(
            "CREATE TRIGGER IF NOT EXISTS order_created_trigger
             AFTER INSERT ON orders
             BEGIN
                 INSERT INTO workflow_events (order_id, event_type, payload)
                 VALUES (NEW.id, 'order_created',
                     json_object('order_id', NEW.id, 'amount', NEW.amount)
                 );


                 SELECT NOTIFY('order_created',
                     json_object('order_id', NEW.id, 'amount', NEW.amount)
                 );
             END",
            [],
        )?;


        Ok(())
    }


    // Start workflow step listeners
    pub async fn start(self: Arc<Self>) -> Result<(), Box<dyn std::error::Error>> {
        let mut tasks = JoinSet::new();


        // Step 1: Listen for new orders, process payment
        let engine = Arc::clone(&self);
        tasks.spawn(async move {
            engine.payment_processor().await
        });


        // Step 2: Listen for payment completion, allocate inventory
        let engine = Arc::clone(&self);
        tasks.spawn(async move {
            engine.inventory_allocator().await
        });


        // Step 3: Listen for inventory allocation, create shipment
        let engine = Arc::clone(&self);
        tasks.spawn(async move {
            engine.shipment_creator().await
        });


        // Wait for all tasks
        while let Some(result) = tasks.join_next().await {
            result??;
        }


        Ok(())
    }


    async fn payment_processor(&self) -> Result<(), Box<dyn std::error::Error>> {
        let subscription = self.pubsub.subscribe("order_created")?;


        loop {
            let notifications = subscription.poll()?;


            for notification in notifications {
                let payload: serde_json::Value = serde_json::from_str(&notification.payload)?;
                let order_id = payload["order_id"].as_i64().unwrap();


                println!("Processing payment for order {}", order_id);


                // Simulate payment processing
                tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;


                // Update order status
                self.db.execute(
                    "UPDATE orders SET status = 'payment_completed' WHERE id = ?1",
                    [order_id.to_string()],
                )?;


                // Notify next step
                self.pubsub.notify(
                    "payment_completed",
                    &serde_json::json!({"order_id": order_id}).to_string(),
                )?;
            }


            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        }
    }


    async fn inventory_allocator(&self) -> Result<(), Box<dyn std::error::Error>> {
        let subscription = self.pubsub.subscribe("payment_completed")?;


        loop {
            let notifications = subscription.poll()?;


            for notification in notifications {
                let payload: serde_json::Value = serde_json::from_str(&notification.payload)?;
                let order_id = payload["order_id"].as_i64().unwrap();


                println!("Allocating inventory for order {}", order_id);


                // Simulate inventory allocation
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;


                // Update order status
                self.db.execute(
                    "UPDATE orders SET status = 'inventory_allocated' WHERE id = ?1",
                    [order_id.to_string()],
                )?;


                // Notify next step
                self.pubsub.notify(
                    "inventory_allocated",
                    &serde_json::json!({"order_id": order_id}).to_string(),
                )?;
            }


            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        }
    }


    async fn shipment_creator(&self) -> Result<(), Box<dyn std::error::Error>> {
        let subscription = self.pubsub.subscribe("inventory_allocated")?;


        loop {
            let notifications = subscription.poll()?;


            for notification in notifications {
                let payload: serde_json::Value = serde_json::from_str(&notification.payload)?;
                let order_id = payload["order_id"].as_i64().unwrap();


                println!("Creating shipment for order {}", order_id);


                // Simulate shipment creation
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;


                // Update order status
                self.db.execute(
                    "UPDATE orders SET status = 'shipped' WHERE id = ?1",
                    [order_id.to_string()],
                )?;


                println!("Order {} workflow complete!", order_id);
            }


            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        }
    }
}


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = Arc::new(WorkflowEngine::new("/data/workflow.db")?);


    // Start workflow engine
    engine.start().await?;


    Ok(())
}

Workflow Architecture:

Order Created (INSERT)
    ↓
Trigger fires NOTIFY 'order_created'
    ↓
Payment Processor (LISTEN 'order_created')
    ↓ (process payment)
    ↓ NOTIFY 'payment_completed'
    ↓
Inventory Allocator (LISTEN 'payment_completed')
    ↓ (allocate inventory)
    ↓ NOTIFY 'inventory_allocated'
    ↓
Shipment Creator (LISTEN 'inventory_allocated')
    ↓ (create shipment)
    ↓ Order Complete

Design targets (not measured; feature planned):

  • Workflow latency: 200-300ms end-to-end
  • No external message broker required
  • Automatic retry on failure (reprocess on startup)
  • Container deployment: Single service

Example 5: Microservices Event Bus - Edge/IoT Deployment

Scenario: IoT sensor network with 100 edge devices coordinating via lightweight event bus for distributed alerting without cloud connectivity.

Edge Device Configuration (edge-config.toml):

[database]
path = "/var/lib/iot/sensors.db"
memory_limit_mb = 64        # Low memory for edge devices
page_size = 512
enable_wal = true


[pubsub]
enabled = true
max_channels = 50
max_payload_bytes = 1024    # Smaller payloads for IoT


[edge]
device_id = "sensor_001"
location = "warehouse_a"

Edge Device Application (Rust):

use heliosdb_nano::{Connection, protocols::PubSubManager};
use std::sync::Arc;
use serde::{Deserialize, Serialize};


#[derive(Debug, Serialize, Deserialize)]
struct SensorReading {
    device_id: String,
    sensor_type: String,
    value: f64,
    timestamp: i64,
    alert: bool,
}


pub struct EdgeSensorNode {
    db: Arc<Connection>,
    pubsub: Arc<PubSubManager>,
    device_id: String,
}


impl EdgeSensorNode {
    pub fn new(device_id: String) -> Result<Self, Box<dyn std::error::Error>> {
        let db = Arc::new(Connection::open("/var/lib/iot/sensors.db")?);
        let pubsub = Arc::new(PubSubManager::new());


        Self::init_schema(&db)?;


        Ok(Self { db, pubsub, device_id })
    }


    fn init_schema(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
        conn.execute(
            "CREATE TABLE IF NOT EXISTS sensor_readings (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                device_id TEXT NOT NULL,
                sensor_type TEXT NOT NULL,
                value REAL NOT NULL,
                alert BOOLEAN DEFAULT 0,
                timestamp INTEGER DEFAULT (strftime('%s', 'now'))
            )",
            [],
        )?;


        // Trigger for critical alerts
        conn.execute(
            "CREATE TRIGGER IF NOT EXISTS critical_alert_trigger
             AFTER INSERT ON sensor_readings
             WHEN NEW.alert = 1
             BEGIN
                 SELECT NOTIFY('critical_alerts',
                     json_object(
                         'device_id', NEW.device_id,
                         'sensor_type', NEW.sensor_type,
                         'value', NEW.value,
                         'timestamp', NEW.timestamp
                     )
                 );
             END",
            [],
        )?;


        Ok(())
    }


    // Record sensor reading with automatic alert detection
    pub fn record_reading(
        &self,
        sensor_type: &str,
        value: f64,
        threshold: f64,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let alert = value > threshold;


        self.db.execute(
            "INSERT INTO sensor_readings (device_id, sensor_type, value, alert)
             VALUES (?1, ?2, ?3, ?4)",
            [
                &self.device_id,
                sensor_type,
                &value.to_string(),
                if alert { "1" } else { "0" },
            ],
        )?;


        // Trigger fires automatically if alert = 1
        Ok(())
    }


    // Listen for alerts from other devices in the network
    pub async fn alert_listener(&self) -> Result<(), Box<dyn std::error::Error>> {
        let subscription = self.pubsub.subscribe("critical_alerts")?;


        loop {
            let notifications = subscription.poll()?;


            for notification in notifications {
                let alert: serde_json::Value = serde_json::from_str(&notification.payload)?;


                println!(
                    "ALERT from {}: {} = {} (threshold exceeded)",
                    alert["device_id"],
                    alert["sensor_type"],
                    alert["value"]
                );


                // Take local action (e.g., activate safety system)
                self.handle_alert(&alert)?;
            }


            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        }
    }


    fn handle_alert(&self, alert: &serde_json::Value) -> Result<(), Box<dyn std::error::Error>> {
        // Local alert handling logic
        // Example: Log alert, activate local safety measures
        println!("Local device {} taking action on alert", self.device_id);
        Ok(())
    }
}


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let node = Arc::new(EdgeSensorNode::new("sensor_001".to_string())?);


    // Start alert listener
    let alert_node = Arc::clone(&node);
    tokio::spawn(async move {
        alert_node.alert_listener().await
    });


    // Simulate sensor readings
    loop {
        let temperature = 20.0 + (rand::random::<f64>() * 10.0);


        // Record reading (triggers alert if temperature > 25.0)
        node.record_reading("temperature", temperature, 25.0)?;


        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    }
}

Edge Network Architecture:

┌────────────────────┐  ┌────────────────────┐  ┌────────────────────┐
│  Edge Device 1     │  │  Edge Device 2     │  │  Edge Device 3     │
│  ┌──────────────┐  │  │  ┌──────────────┐  │  │  ┌──────────────┐  │
│  │ Sensor Data  │  │  │  │ Sensor Data  │  │  │  │ Sensor Data  │  │
│  └──────────────┘  │  │  └──────────────┘  │  │  └──────────────┘  │
│  ┌──────────────┐  │  │  ┌──────────────┐  │  │  ┌──────────────┐  │
│  │ HeliosDB Nano│  │  │  │ HeliosDB Nano│  │  │  │ HeliosDB Nano│  │
│  │ + PubSub     │  │  │  │ + PubSub     │  │  │  │ + PubSub     │  │
│  └──────────────┘  │  │  └──────────────┘  │  │  └──────────────┘  │
│  LISTEN 'alerts'   │  │  LISTEN 'alerts'   │  │  LISTEN 'alerts'   │
│  NOTIFY on alert   │  │  NOTIFY on alert   │  │  NOTIFY on alert   │
└────────────────────┘  └────────────────────┘  └────────────────────┘
           │                      │                      │
           └──────────────────────┴──────────────────────┘
                   Local Event Propagation
                   (in-memory, sub-millisecond)

Design targets (not measured; feature planned):

  • Alert propagation: <1ms within process
  • Memory footprint: 32MB per device
  • No cloud dependency (fully offline)
  • Scales to hundreds of devices in local network

Technical Advantages

How the Planned Design Compares

The HeliosDB Nano column describes the planned design. Its latency and memory cells are design targets, not measurements.

AspectHeliosDB Nano (planned)Redis Pub/SubPostgreSQL LISTEN/NOTIFYKafka/RabbitMQ
Notification LatencyIn-process delivery, no network hop (target)5-20ms (network)10-50ms (network)50-200ms (network)
Memory FootprintIn-memory channel registry (target: small, per subscription)200MB+500MB+1GB+
Deployment ComplexitySingle binarySeparate serviceSeparate serviceMultiple services
Infrastructure Cost$0 (embedded)$100-500/month$50-200/month$500-2000/month
Offline CapabilityFull supportNoNoNo
Database IntegrationNative (triggers)ManualNative (triggers)Manual
Transaction SupportACID-compliantNoACID-compliantNo
Edge Device ViabilityYes (planned)NoNoNo

Key Success Metrics

Technical KPIs

MetricTargetMeasurement Method
Notification Latency (P99)<5msPrometheus metrics: histogram(notification_delivery_duration_ms)
Notification Throughput10K+ msgs/secPrometheus counter: rate(notifications_delivered_total[1m])
Memory Overhead<10MB for 1000 channelsProcess RSS measurement: process.memory.rss
Subscription LifecycleZero leaksMonitor subscription count: active_subscriptions should match LISTEN commands
Delivery Reliability100%Compare notifications_sent to notifications_received

Business KPIs

MetricTargetMeasurement Method
Infrastructure Cost Reduction80-95%Monthly bill comparison: (Redis cost before - $0 after) / Redis cost before
CPU Efficiency Gain20-40% reductionApplication CPU metrics: (polling CPU - event-driven CPU) / polling CPU
Real-Time Feature Latency<100ms end-to-endUser-facing metrics: time from data change to UI update
Developer Productivity2-5x faster feature developmentTime to implement real-time feature: polling vs LISTEN/NOTIFY
Operational Complexity70% reductionCount of deployed services: before (DB + Redis + app) vs after (app only)

Conclusion

Real-time event-driven applications, from live dashboards and collaborative tools to cache invalidation and workflow automation, usually choose between an external message broker (Redis, Kafka) that adds a network hop and another service to operate, and polling that spends CPU and delays updates. PostgreSQL-compatible LISTEN/NOTIFY is planned for HeliosDB Nano to offer a third option: in-process notifications with no external dependency, delivered only after COMMIT. It is not in the current release, and the figures on this page are design targets, not measurements.

Next Steps: until LISTEN/NOTIFY is available, poll for changes, for example with an indexed updated_at or sequence column that clients read past their last seen value, and keep an external broker where you need push delivery today. See Known Limitations for the current status, and the Nano changelog for the release that adds it.

References

  1. PostgreSQL LISTEN/NOTIFY Documentation: https://www.postgresql.org/docs/current/sql-notify.html
  2. Redis Pub/Sub Performance Benchmarks: https://redis.io/docs/manual/pubsub/
  3. “The Database Poll Pattern” Anti-Pattern Analysis: Martin Fowler, Enterprise Integration Patterns
  4. Edge Computing Market Size: IDC IoT and Edge Computing Forecast 2024-2028
  5. Real-Time Web Application Architecture: “Designing Data-Intensive Applications” by Martin Kleppmann
  6. Cache Invalidation Strategies: “Web Application Performance” (O’Reilly, 2023)

Review Cycle: Quarterly Owner: Product Marketing Adapted for: HeliosDB Nano Embedded Database