feature: tap rollout to replace jetstream

.gitignore

···
 /keys.json
 /.vscode/launch.json
 .*.env
-.env
+.envtmp/

CLAUDE.md

···
 
 ## Project Overview
 
-Smokesignal is a Rust-based event and RSVP management application built for the AT Protocol ecosystem. It provides decentralized identity management, OAuth authentication, event coordination, real-time event streaming, search capabilities, webhook notifications, and email notifications. The application features RSVP acceptance workflows, private event content with conditional display, profile caching, and content storage with S3 support. It supports both standard AT Protocol OAuth (PDS) and AIP (AT Protocol Improvement Proposal) OAuth flows, with backend selection determined by runtime configuration.
+Smokesignal is a Rust-based event and RSVP management application built for the AT Protocol ecosystem. It provides decentralized identity management, OAuth authentication, event coordination, real-time event streaming, search capabilities, and email notifications. The application features RSVP acceptance workflows, private event content with conditional display, profile caching, and content storage with S3 support. It supports both standard AT Protocol OAuth (PDS) and AIP (AT Protocol Improvement Proposal) OAuth flows, with backend selection determined by runtime configuration.
 
 ## Tech Stack
 
···
 - **Cryptography**: P-256 ECDSA with elliptic-curve and p256 crates
 - **Task Management**: Tokio with task tracking and graceful shutdown
 - **DNS Resolution**: Hickory resolver with DoH/DoT support
-- **Real-time**: Jetstream consumer for AT Protocol firehose
-- **Webhooks**: Event notification system with background processing
+- **Real-time**: TAP consumer for AT Protocol events with backfill support
 - **Email**: Lettre for SMTP-based email notifications
 - **Image Processing**: Image resizing and optimization for avatars and content
 - **Rate Limiting**: Redis-backed throttling for API endpoints
···
 │   ├── denylist.rs   # Handle blocking
 │   ├── content.rs    # Content storage abstraction (filesystem/S3)
 │   ├── private_event_content.rs # Private event content
-│   ├── webhook.rs    # Webhook storage operations
 │   ├── notification.rs # Email notification preferences
 │   ├── distributed_lock.rs # Redis-based distributed locking
 │   └── atproto.rs    # AT Protocol specific storage
 ├── config.rs         # Environment configuration
 ├── key_provider.rs   # JWT key management
 ├── i18n.rs          # Internationalization
-├── consumer.rs       # Jetstream event consumer
+├── tap_processor.rs  # TAP event consumer and processor
 ├── processor.rs      # Content fetcher for events
 ├── service.rs        # Service document and DID management
-├── webhooks.rs       # Webhook event constants
 ├── emailer.rs        # Email sending functionality
 ├── email_templates.rs # Email template rendering
 ├── email_confirmation.rs # Email confirmation tokens
···
 ├── facets.rs         # AT Protocol facets parsing
 ├── task_identity_refresh.rs # Identity refresh task
 ├── task_oauth_requests_cleanup.rs # OAuth cleanup task
-├── task_search_indexer.rs # OpenSearch indexing
-└── task_webhooks.rs  # Webhook processing
+└── task_search_indexer.rs # OpenSearch indexing
 ```
 
 ### Key Patterns
···
 - **Dependency Injection**: Services passed through application context
 - **Template-driven UI**: Server-side rendering with optional reloading
 - **Background Tasks**: Multiple async workers for maintenance and processing
-- **Event-driven**: Jetstream consumer for real-time AT Protocol events
+- **Event-driven**: TAP consumer for real-time AT Protocol events with backfill support
 
 ### Database Schema
 - `identity_profiles` - User identity and preferences (including discovery settings)
···
 - `private_event_content` - Conditional event content based on display criteria
 - `denylist` - Handle blocking for moderation
 - `did_documents` - Cached DID documents for performance
-- `identity_webhooks` - User-configured webhook endpoints
 
 ## Configuration Requirements
 
···
 - `RUST_LOG` - Logging configuration (default: info)
 - `PORT` - Server port (default: 3000)
 - `BIND_ADDR` - Bind address (default: 0.0.0.0)
-- `ENABLE_JETSTREAM` - Enable Jetstream consumer (default: true)
-- `ENABLE_WEBHOOKS` - Enable webhook functionality
-- `ENABLE_TASK_WEBHOOKS` - Enable webhook processing task
+- `TAP_HOSTNAME` - TAP service hostname (default: localhost:2480)
+- `TAP_PASSWORD` - TAP admin password for authentication (optional)
+- `ENABLE_TAP` - Enable TAP consumer (default: true)
 - `ENABLE_OPENSEARCH` - Enable OpenSearch integration
 - `ENABLE_TASK_OPENSEARCH` - Enable search indexing task
 - `OPENSEARCH_ENDPOINT` - OpenSearch server endpoint
···
 - Internationalization support with Fluent
 - LRU caching for OAuth requests and DID documents
 - Graceful shutdown with task tracking and cancellation tokens
-- Real-time event ingestion via Jetstream
+- Real-time event ingestion via TAP with backfill support
 - Full-text search with OpenSearch
-- Webhook notifications for events and RSVPs
 - Email notifications with HTML templates and confirmation workflow
 - RSVP acceptance/validation workflow with tickets and records
 - Private event content with conditional display criteria
···
   - Profiles (events.smokesignal.profile)
   - Acceptance records (events.smokesignal.calendar.acceptance)
   - Location data (community.lexicon.location)
-- Jetstream consumer for real-time event ingestion from AT Protocol firehose
+- TAP consumer for real-time event ingestion from AT Protocol with backfill
 - XRPC server endpoints for search and service operations
 - Service identity with did:web support for federated interactions
 - AT Protocol attestation support for verified claims
···
   - `atproto-oauth-aip` - AIP OAuth implementation
   - `atproto-client` - AT Protocol client functionality
   - `atproto-record` - Record management
-  - `atproto-jetstream` - Firehose event streaming
+  - `atproto-tap` - TAP event streaming with backfill support
   - `atproto-xrpcs` - XRPC server implementation
   - `atproto-attestation` - Attestation verification
 

Cargo.toml

···
 atproto-client = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs" }
 atproto-attestation = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs" }
 atproto-identity = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs", features = ["lru", "zeroize", "hickory-dns"] }
-atproto-jetstream = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs" }
+atproto-tap = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs" }
 atproto-oauth = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs", features = ["lru", "zeroize"] }
 atproto-oauth-aip = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs" }
 atproto-oauth-axum = { git = "https://tangled.org/@smokesignal.events/atproto-identity-rs", features = ["zeroize"] }
···
 sqlx = { version = "0.8", default-features = false, features = ["derive", "macros", "migrate", "json", "runtime-tokio", "postgres", "chrono", "tls-rustls-ring-native-roots"] }
 thiserror = "2.0"
 tokio = { version = "1.41", features = ["bytes", "macros", "net", "rt", "rt-multi-thread", "signal", "sync"] }
+tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["net", "rt", "tracing"] }
 tower-http = { version = "0.6", features = ["cors", "fs", "timeout", "trace", "tracing"] }
 tracing = { version = "0.1", features = ["async-await", "log", "valuable"] }
···
 url = "2.5"
 urlencoding = "2.1"
 base32 = "0.5.1"
+futures = { version = "0.3.31", default-features = false, features = ["alloc"] }
 
 [profile.release]
 opt-level = 3

migrations/20251225133302_drop_identity_webhooks.sql

···
+-- Drop identity_webhooks table (webhook feature removed)
+DROP TABLE IF EXISTS identity_webhooks;

src/bin/smokesignal.rs

···
 use atproto_identity::resolve::{
     HickoryDnsResolver, InnerIdentityResolver, SharedIdentityResolver,
 };
-use atproto_jetstream::{CancellationToken, Consumer as JetstreamConsumer, ConsumerTaskConfig};
 use atproto_oauth_axum::state::OAuthClientConfig;
-use smokesignal::consumer::Consumer;
 use smokesignal::processor::ContentFetcher;
+use smokesignal::tap_processor::TapProcessor;
+use smokesignal::task_search_indexer::SearchIndexer;
+use tokio_util::sync::CancellationToken;
 use smokesignal::service::{ServiceDID, ServiceKey, build_service_document};
 use smokesignal::storage::content::{CachedContentStorage, ContentStorage, FilesystemStorage};
 use smokesignal::{
···
     storage::{
         atproto::{PostgresDidDocumentStorage, PostgresOAuthRequestStorage},
         cache::create_cache_pool,
-        distributed_lock::DistributedLock,
     },
 };
 
···
 use std::{collections::HashMap, env, str::FromStr, sync::Arc};
 use tokio::net::TcpListener;
 use tokio::signal;
-use tokio::sync::mpsc;
 use tokio_util::task::TaskTracker;
 use tracing_subscriber::prelude::*;
 use unic_langid::LanguageIdentifier;
···
         plc_hostname: config.plc_hostname.clone(),
     })));
 
-    // Create webhook channel if webhooks are enabled
-    let webhook_sender = if config.enable_webhooks && config.enable_task_webhooks {
-        let (sender, receiver) = mpsc::channel(100);
-        Some((sender, receiver))
-    } else {
-        None
-    };
-
     let content_storage: Arc<dyn ContentStorage> = if config.content_storage.starts_with("s3://") {
         #[cfg(feature = "s3")]
         {
···
         supported_languages,
         locales,
         content_storage.clone(),
-        webhook_sender.as_ref().map(|(sender, _)| sender.clone()),
         emailer,
         service_did,
         service_document,
···
         });
     }
 
-    if config.enable_jetstream {
-        // Try to acquire distributed lock for Jetstream consumer
-        tracing::info!("Attempting to acquire Jetstream consumer lock...");
-        match DistributedLock::new(cache_pool.clone()).await {
-            Ok(mut distributed_lock) => {
-                // Try to acquire the lock with retry for up to 60 seconds
-                // This is longer than the lock TTL (30s) to ensure any stale locks
-                // from crashed instances will expire before we give up
-                let acquired = match distributed_lock
-                    .acquire_with_retry(std::time::Duration::from_secs(60))
-                    .await
-                {
-                    Ok(acquired) => acquired,
-                    Err(e) => {
-                        tracing::error!("Failed to acquire distributed lock: {}", e);
-                        false
-                    }
-                };
+    if config.enable_tap {
+        // Create resolvers for signature verification
+        let record_resolver = Arc::new(
+            smokesignal::record_resolver::StorageBackedRecordResolver::new(
+                http_client.clone(),
+                identity_resolver.clone(),
+                pool.clone(),
+            ),
+        );
+        let key_resolver = IdentityDocumentKeyResolver::new(identity_resolver.clone());
 
-                if !acquired {
-                    tracing::warn!(
-                        "Could not acquire Jetstream consumer lock after 60 seconds - another instance may be running"
-                    );
-                    tracing::info!("This instance will not consume Jetstream events");
-                    tracing::info!("If no other instance is running, the lock will expire in up to 30 seconds");
-                } else {
-                    tracing::debug!(
-                        "Successfully acquired Jetstream consumer lock - starting event consumption"
-                    );
-
-                    // Start lock renewal task
-                    let lock_renewal_token = token.clone();
-                    let lock_renewal_handle =
-                        distributed_lock.start_renewal_task(lock_renewal_token);
-
-                    // Track the renewal task
-                    tracker.spawn(lock_renewal_handle);
-
-                    // Now proceed with normal Jetstream consumer setup
-                    let consumer = Consumer {};
-                    let (smoke_signal_event_handler, event_receiver) =
-                        consumer.create_handler("content_fetcher");
-
-                    // Create a second handler for the search indexer if enabled
-                    let (search_indexer_handler, search_indexer_receiver) =
-                        if config.enable_opensearch && config.enable_task_opensearch {
-                            let (handler, receiver) = consumer.create_handler("search_indexer");
-                            (Some(handler), Some(receiver))
-                        } else {
-                            (None, None)
-                        };
+        // Create content fetcher
+        let content_fetcher = ContentFetcher::new(
+            pool.clone(),
+            content_storage.clone(),
+            identity_resolver.clone(),
+            document_storage.clone(),
+            http_client.clone(),
+            record_resolver,
+            key_resolver,
+        );
 
-                    // Create resolvers for signature verification
-                    let record_resolver = Arc::new(
-                        smokesignal::record_resolver::StorageBackedRecordResolver::new(
-                            http_client.clone(),
-                            identity_resolver.clone(),
-                            pool.clone(),
-                        ),
-                    );
-                    let key_resolver = IdentityDocumentKeyResolver::new(identity_resolver.clone());
+        // Create search indexer if enabled
+        let search_indexer = if config.enable_opensearch && config.enable_task_opensearch {
+            let opensearch_endpoint = config.opensearch_endpoint.as_ref().expect(
+                "OPENSEARCH_ENDPOINT is required when search indexing is enabled",
+            );
 
-                    let content_fetcher = ContentFetcher::new(
-                        pool.clone(),
-                        content_storage.clone(),
-                        identity_resolver.clone(),
-                        document_storage.clone(),
-                        http_client.clone(),
-                        record_resolver,
-                        key_resolver,
-                    );
+            match SearchIndexer::new(
+                opensearch_endpoint,
+                identity_resolver.clone(),
+                document_storage.clone(),
+            )
+            .await
+            {
+                Ok(indexer) => Some(indexer),
+                Err(err) => {
+                    tracing::error!("Failed to create search indexer: {}", err);
+                    None
+                }
+            }
+        } else {
+            None
+        };
 
-                    let inner_token = token.clone();
-                    tracker.spawn(async move {
-                        tokio::select! {
-                            result = content_fetcher.start_processing(event_receiver) => {
-                                if let Err(err) = result {
-                                    tracing::error!(error = ?err, "failed processing stuff");
-                                }
-                            }
-                            () = inner_token.cancelled() => {
-                                tracing::info!("content processing stopped");
-                            }
-                        }
-                    });
+        // Create and spawn the unified TAP processor
+        // No distributed lock needed - TAP manages the outbox with acks and retries,
+        // so multiple instances can safely consume in parallel
+        let tap_processor = TapProcessor::new(
+            &config.tap_hostname,
+            config.tap_password.clone(),
+            &config.user_agent,
+            content_fetcher,
+            search_indexer,
+            token.clone(),
+        );
 
-                    let inner_token = token.clone();
-                    let inner_config = config.clone();
-                    tracker.spawn(async move {
-                        let mut disconnect_times = Vec::new();
-                        let disconnect_window = std::time::Duration::from_secs(60); // 1 minute window
-                        let max_disconnects_per_minute = 1;
-                        let reconnect_delay = std::time::Duration::from_secs(5);
-
-                        loop {
-                            // Create new consumer for each connection attempt
-                            let jetstream_config = ConsumerTaskConfig {
-                                user_agent: inner_config.user_agent.clone(),
-                                compression: false,
-                                zstd_dictionary_location: String::new(),
-                                jetstream_hostname: inner_config.jetstream_hostname.clone(),
-                                collections: vec![
-                                    "community.lexicon.calendar.rsvp".to_string(),
-                                    "community.lexicon.calendar.event".to_string(),
-                                    "events.smokesignal.profile".to_string(),
-                                    "events.smokesignal.calendar.acceptance".to_string()
-                                    ],
-                                dids: vec![],
-                                max_message_size_bytes: Some(20 * 1024 * 1024), // 10MB
-                                cursor: None,
-                                require_hello: true,
-                            };
-
-                            let jetstream_consumer = JetstreamConsumer::new(jetstream_config);
-
-                            if let Err(err) = jetstream_consumer.register_handler(smoke_signal_event_handler.clone()).await {
-                                tracing::error!("Failed to register handler: {}", err);
-                                inner_token.cancel();
-                                break;
-                            }
-
-                            // Register search indexer handler if enabled
-                            if let Some(search_handler) = &search_indexer_handler
-                                && let Err(err) = jetstream_consumer.register_handler(search_handler.clone()).await {
-                                tracing::error!("Failed to register search indexer handler: {}", err);
-                                inner_token.cancel();
-                                break;
-                            }
-
-                            tokio::select! {
-                                result = jetstream_consumer.run_background(inner_token.clone()) => {
-                                    if let Err(err) = result {
-                                        let now = std::time::Instant::now();
-                                        disconnect_times.push(now);
-
-                                        // Remove disconnect times older than the window
-                                        disconnect_times.retain(|&t| now.duration_since(t) <= disconnect_window);
-
-                                        if disconnect_times.len() > max_disconnects_per_minute {
-                                            tracing::error!(
-                                                "Jetstream disconnect rate exceeded: {} disconnects in 1 minute, exiting",
-                                                disconnect_times.len()
-                                            );
-                                            inner_token.cancel();
-                                            break;
-                                        }
-
-                                        tracing::error!("Jetstream disconnected: {}, reconnecting in {:?}", err, reconnect_delay);
-
-                                        // Wait before reconnecting
-                                        tokio::select! {
-                                            () = tokio::time::sleep(reconnect_delay) => {},
-                                            () = inner_token.cancelled() => {
-                                                tracing::info!("Jetstream consumer cancelled during reconnect delay");
-                                                break;
-                                            }
-                                        }
-
-                                        // Continue the loop to reconnect
-                                        continue;
-                                    }
-                                }
-                                () = inner_token.cancelled() => {
-                                    tracing::info!("Jetstream consumer cancelled");
-                                    break;
-                                }
-                            }
-
-                            // If we reach here, the consumer exited without error (unlikely)
-                            tracing::info!("Jetstream consumer exited normally");
-                            break;
-
-                        }
-                    });
-
-                    // Spawn search indexer task if enabled
-                    if let Some(receiver) = search_indexer_receiver {
-                        use smokesignal::task_search_indexer::SearchIndexer;
-
-                        let opensearch_endpoint = config.opensearch_endpoint.as_ref().expect(
-                            "OPENSEARCH_ENDPOINT is required when search indexing is enabled",
-                        );
-
-                        match SearchIndexer::new(
-                            opensearch_endpoint,
-                            identity_resolver.clone(),
-                            document_storage.clone(),
-                            receiver,
-                            token.clone(),
-                        )
-                        .await
-                        {
-                            Ok(search_indexer) => {
-                                let inner_token = token.clone();
-                                tracker.spawn(async move {
-                                    if let Err(err) = search_indexer.run().await {
-                                        tracing::error!("Search indexer task failed: {}", err);
-                                    }
-                                    inner_token.cancel();
-                                });
-                            }
-                            Err(err) => {
-                                tracing::error!("Failed to create search indexer: {}", err);
-                                token.cancel();
-                            }
-                        }
-                    }
-                }
+        let inner_token = token.clone();
+        tracker.spawn(async move {
+            if let Err(err) = tap_processor.run().await {
+                tracing::error!(error = ?err, "TAP processor failed");
             }
-            Err(e) => {
-                tracing::error!("Failed to create distributed lock: {}", e);
-                tracing::warn!("Jetstream consumer disabled - could not create lock");
-                // Continue without Jetstream if we can't create the lock
-                // This allows the application to still function without event consumption
-            }
-        }
+            inner_token.cancel();
+        });
     }
 
     // Spawn OAuth requests cleanup task if enabled
···
         tracker.spawn(async move {
             if let Err(err) = cleanup_task.run().await {
                 tracing::error!("OAuth requests cleanup task failed: {}", err);
-            }
-            inner_token.cancel();
-        });
-    }
-
-    // Spawn webhook processor task if enabled
-    if let Some((_, receiver)) = webhook_sender {
-        use axum::extract::FromRef;
-        use smokesignal::task_webhooks::WebhookProcessor;
-
-        let webhook_processor = WebhookProcessor::new(
-            pool.clone(),
-            document_storage.clone(),
-            ServiceDID::from_ref(&web_context),
-            ServiceKey::from_ref(&web_context),
-            receiver,
-            token.clone(),
-        );
-
-        let inner_token = token.clone();
-        tracker.spawn(async move {
-            if let Err(err) = webhook_processor.run().await {
-                tracing::error!("Webhook processor task failed: {}", err);
             }
             inner_token.cancel();
         });

src/config.rs

···
     pub user_agent: String,
     pub database_url: String,
     pub plc_hostname: String,
-    pub jetstream_hostname: String,
+    pub tap_hostname: String,
+    pub tap_password: Option<String>,
     pub redis_url: String,
     pub admin_dids: AdminDIDs,
     pub dns_nameservers: DnsNameservers,
     pub oauth_backend: OAuthBackendConfig,
     pub enable_task_oauth_requests_cleanup: bool,
     pub enable_task_identity_refresh: bool,
-    pub enable_jetstream: bool,
+    pub enable_tap: bool,
     pub content_storage: String,
-    pub enable_webhooks: bool,
-    pub enable_task_webhooks: bool,
     pub service_key: ServiceKey,
     pub enable_opensearch: bool,
     pub enable_task_opensearch: bool,
···
 
         let plc_hostname = default_env("PLC_HOSTNAME", "plc.directory");
 
-        let jetstream_hostname =
-            default_env("JETSTREAM_HOSTNAME", "jetstream2.us-east.bsky.network");
+        let tap_hostname = default_env("TAP_HOSTNAME", "localhost:2480");
+        let tap_password_str = optional_env("TAP_PASSWORD");
+        let tap_password = if tap_password_str.is_empty() {
+            None
+        } else {
+            Some(tap_password_str)
+        };
 
         let database_url = default_env("DATABASE_URL", "sqlite://development.db");
 
···
             .parse::<bool>()
             .unwrap_or(true);
 
-        let enable_jetstream = default_env("ENABLE_JETSTREAM", "true")
-            .parse::<bool>()
-            .unwrap_or(true);
+        let enable_tap = parse_bool_env("ENABLE_TAP", true);
 
         let content_storage = require_env("CONTENT_STORAGE")?;
-
-        // Parse webhook enablement flags
-        let enable_webhooks = parse_bool_env("ENABLE_WEBHOOKS", false);
-        let enable_task_webhooks = parse_bool_env("ENABLE_TASK_WEBHOOKS", false);
 
         let service_key: ServiceKey = require_env("SERVICE_KEY")?.try_into()?;
 
···
             certificate_bundles,
             user_agent,
             plc_hostname,
-            jetstream_hostname,
+            tap_hostname,
+            tap_password,
             database_url,
             http_cookie_key,
             redis_url,
···
             oauth_backend,
             enable_task_oauth_requests_cleanup,
             enable_task_identity_refresh,
-            enable_jetstream,
+            enable_tap,
             content_storage,
-            enable_webhooks,
-            enable_task_webhooks,
             service_key,
             enable_opensearch,
             enable_task_opensearch,

src/consumer.rs

···
-use anyhow::Result;
-use async_trait::async_trait;
-use atproto_jetstream::{EventHandler, JetstreamEvent};
-use std::sync::Arc;
-use tokio::sync::mpsc;
-
-use crate::atproto::lexicon::{acceptance::NSID as ACCEPTANCE_NSID, profile::NSID as PROFILE_NSID};
-use atproto_record::lexicon::community::lexicon::{
-    calendar::event::NSID as EVENT_NSID, calendar::rsvp::NSID as RSVP_NSID,
-};
-
-pub type SmokeSignalEventReceiver = mpsc::UnboundedReceiver<SmokeSignalEvent>;
-
-#[derive(Debug, Clone)]
-pub enum SmokeSignalEvent {
-    Commit {
-        did: String,
-        collection: String,
-        rkey: String,
-        cid: String,
-        record: serde_json::Value,
-    },
-    Delete {
-        did: String,
-        collection: String,
-        rkey: String,
-    },
-}
-
-pub struct SmokeSignalEventHandler {
-    event_sender: mpsc::UnboundedSender<SmokeSignalEvent>,
-    handler_id: String,
-}
-
-impl SmokeSignalEventHandler {
-    fn new(event_sender: mpsc::UnboundedSender<SmokeSignalEvent>, handler_id: String) -> Self {
-        Self {
-            event_sender,
-            handler_id,
-        }
-    }
-}
-
-#[async_trait]
-impl EventHandler for SmokeSignalEventHandler {
-    async fn handle_event(&self, event: JetstreamEvent) -> Result<()> {
-        let incoming_event = match event {
-            JetstreamEvent::Commit { did, commit, .. } => {
-                if commit.collection != RSVP_NSID
-                    && commit.collection != EVENT_NSID
-                    && commit.collection != PROFILE_NSID
-                    && commit.collection != ACCEPTANCE_NSID
-                {
-                    return Ok(());
-                }
-
-                SmokeSignalEvent::Commit {
-                    did,
-                    collection: commit.collection,
-                    rkey: commit.rkey,
-                    cid: commit.cid,
-                    record: commit.record,
-                }
-            }
-            JetstreamEvent::Delete { did, commit, .. } => {
-                if commit.collection != RSVP_NSID
-                    && commit.collection != EVENT_NSID
-                    && commit.collection != PROFILE_NSID
-                    && commit.collection != ACCEPTANCE_NSID
-                {
-                    return Ok(());
-                }
-                SmokeSignalEvent::Delete {
-                    did,
-                    collection: commit.collection,
-                    rkey: commit.rkey,
-                }
-            }
-            JetstreamEvent::Identity { .. } | JetstreamEvent::Account { .. } => {
-                return Ok(());
-            }
-        };
-
-        if let Err(err) = self.event_sender.send(incoming_event) {
-            tracing::error!(error = ?err, "error sending smokesignal event to processing");
-        }
-
-        Ok(())
-    }
-
-    fn handler_id(&self) -> String {
-        self.handler_id.clone()
-    }
-}
-
-pub struct Consumer {}
-
-impl Consumer {
-    pub fn create_handler(
-        &self,
-        handler_id: &str,
-    ) -> (Arc<SmokeSignalEventHandler>, SmokeSignalEventReceiver) {
-        let (sender, receiver) = mpsc::unbounded_channel();
-        let handler = Arc::new(SmokeSignalEventHandler::new(sender, handler_id.to_string()));
-        (handler, receiver)
-    }
-}

src/http/context.rs

···
 use minijinja::context as template_context;
 use std::convert::Infallible;
 use std::{ops::Deref, sync::Arc};
-use tokio::sync::mpsc;
 use unic_langid::LanguageIdentifier;
 
 #[cfg(all(feature = "reload", not(feature = "embed")))]
···
 use crate::mcp::session::McpSessionManager;
 use crate::service::{ServiceDID, ServiceDocument, ServiceKey};
 use crate::storage::content::ContentStorage;
-use crate::task_webhooks::TaskWork;
 use crate::{
     config::Config,
     http::middleware_auth::Auth,
···
     pub(crate) oauth_storage: Arc<dyn OAuthRequestStorage>,
     pub(crate) document_storage: Arc<dyn DidDocumentStorage>,
     pub(crate) content_storage: Arc<dyn ContentStorage>,
-    pub(crate) webhook_sender: Option<mpsc::Sender<TaskWork>>,
     pub(crate) emailer: Option<Arc<dyn Emailer>>,
     pub(crate) service_did: ServiceDID,
     pub(crate) service_document: ServiceDocument,
···
         supported_languages: Vec<LanguageIdentifier>,
         locales: Locales,
         content_storage: Arc<dyn ContentStorage>,
-        webhook_sender: Option<mpsc::Sender<TaskWork>>,
         emailer: Option<Arc<dyn Emailer>>,
         service_did: ServiceDID,
         service_document: ServiceDocument,
···
             oauth_storage,
             document_storage,
             content_storage,
-            webhook_sender,
             emailer,
             service_did,
             service_document,

src/http/handle_create_event.rs

···
 use crate::http::utils::url_from_aturi;
 use crate::select_template;
 use crate::storage::event::{EventInsertParams, event_insert_with_metadata};
-use crate::storage::webhook::webhook_list_enabled_by_did;
-use crate::task_webhooks::TaskWork;
 use atproto_client::com::atproto::repo::{
     CreateRecordRequest, CreateRecordResponse, create_record,
 };
···
                     }
                 }
 
-                // Send webhooks if enabled
-                if web_context.config.enable_webhooks
-                    && let Some(webhook_sender) = &web_context.webhook_sender
-                {
-                    // Get all enabled webhooks for the user
-                    if let Ok(webhooks) =
-                        webhook_list_enabled_by_did(&web_context.pool, &current_handle.did).await
-                    {
-                        // Prepare context with email if shared
-                        let context = json!({});
-
-                        let record_json = json!({
-                            "uri": &create_record_response.uri,
-                            "cit": &create_record_response.cid,
-                        });
-
-                        // Send webhook for each enabled webhook
-                        for webhook in webhooks {
-                            let _ = webhook_sender
-                                .send(TaskWork::EventCreated {
-                                    identity: current_handle.did.clone(),
-                                    service: webhook.service,
-                                    record: record_json.clone(),
-                                    context: context.clone(),
-                                })
-                                .await;
-                        }
-                    }
-                }
-
                 let event_url = url_from_aturi(
                     &web_context.config.external_base,
                     &create_record_response.uri,
···
             );
             // Don't fail the request - the event was created successfully
             // The header can be fetched later via jetstream or admin import
-        }
-    }
-
-    // Send webhooks if enabled
-    if web_context.config.enable_webhooks
-        && let Some(webhook_sender) = &web_context.webhook_sender
-    {
-        if let Ok(webhooks) =
-            webhook_list_enabled_by_did(&web_context.pool, &current_handle.did).await
-        {
-            let context = json!({});
-            let record_json = json!({
-                "uri": &create_record_response.uri,
-                "cid": &create_record_response.cid,
-            });
-
-            for webhook in webhooks {
-                let _ = webhook_sender
-                    .send(TaskWork::EventCreated {
-                        identity: current_handle.did.clone(),
-                        service: webhook.service,
-                        record: record_json.clone(),
-                        context: context.clone(),
-                    })
-                    .await;
-            }
         }
     }
 

src/http/handle_create_rsvp.rs

···
     },
     select_template,
     storage::event::{RsvpInsertParams, rsvp_get_by_event_and_did, rsvp_insert_with_metadata},
-    storage::webhook::webhook_list_enabled_by_did,
-    task_webhooks::TaskWork,
 };
 use atproto_client::com::atproto::repo::{PutRecordRequest, PutRecordResponse, put_record};
 use atproto_record::lexicon::community::lexicon::calendar::rsvp::{NSID, Rsvp, RsvpStatus};
···
                                     }
                                 }
                             }
-                        }
-                    }
-                }
-
-                // Send webhooks if enabled
-                if web_context.config.enable_webhooks
-                    && let Some(webhook_sender) = &web_context.webhook_sender
-                {
-                    let webhook_identity =
-                        ATURI::from_str(build_rsvp_form.subject_aturi.as_ref().unwrap())
-                            .map(|value| value.authority)
-                            .unwrap_or_default();
-
-                    // Get all enabled webhooks for the user
-                    if let Ok(webhooks) =
-                        webhook_list_enabled_by_did(&web_context.pool, &webhook_identity).await
-                    {
-                        // Prepare context (empty - email sharing removed)
-                        let context = serde_json::json!({});
-
-                        // Convert the RSVP record to JSON
-                        let record_json = serde_json::json!({
-                            "uri": &create_record_result.uri,
-                            "cid": &create_record_result.cid,
-                        });
-
-                        // Send webhook for each enabled webhook
-                        for webhook in webhooks {
-                            let _ = webhook_sender
-                                .send(TaskWork::RSVPCreated {
-                                    identity: current_handle.did.clone(),
-                                    service: webhook.service,
-                                    record: record_json.clone(),
-                                    context: context.clone(),
-                                })
-                                .await;
                         }
                     }
                 }

src/http/handle_settings.rs

···
             notification_get, notification_reset_confirmation, notification_set_preference,
         },
         profile::{profile_get_by_did, profile_insert},
-        webhook::{webhook_delete, webhook_list_by_did, webhook_toggle_enabled, webhook_upsert},
     },
-    task_webhooks::TaskWork,
-    webhooks::SMOKE_SIGNAL_AUTOMATION_SERVICE,
 };
 
 #[derive(Deserialize, Clone, Debug)]
···
 #[derive(Deserialize, Clone, Debug)]
 pub(crate) struct EmailForm {
     email: Option<String>,
-}
-
-#[derive(Deserialize, Clone, Debug)]
-pub(crate) struct WebhookForm {
-    service: String,
 }
 
 #[derive(Deserialize, Clone, Debug)]
···
         .map(|lang| lang.to_string())
         .collect::<Vec<String>>();
 
-    // Get webhooks if enabled
-    let webhooks = if web_context.config.enable_webhooks {
-        webhook_list_by_did(&web_context.pool, &current_handle.did).await?
-    } else {
-        vec![]
-    };
-
     // Get profile data if it exists
     let profile_record = profile_get_by_did(&web_context.pool, &current_handle.did).await?;
     let (profile, profile_display_name, profile_description, profile_host) = if let Some(prof_rec) =
···
             template_context! {
                 timezones => timezones,
                 languages => supported_languages,
-                webhooks => webhooks,
-                webhooks_enabled => web_context.config.enable_webhooks,
                 profile,
                 profile_display_name,
                 profile_description,
···
                 current_handle,
                 email_updated => true,
                 ..default_context
-            },
-        ),
-    )
-        .into_response())
-}
-
-pub(crate) async fn handle_add_webhook(
-    State(web_context): State<WebContext>,
-    identity_resolver: State<Arc<dyn IdentityResolver>>,
-    Language(language): Language,
-    Cached(auth): Cached<Auth>,
-    Form(webhook_form): Form<WebhookForm>,
-) -> Result<impl IntoResponse, WebError> {
-    // Check if webhooks are enabled
-    if !web_context.config.enable_webhooks {
-        return Ok((StatusCode::NOT_FOUND, "Not found").into_response());
-    }
-
-    let current_handle = auth.require_flat()?;
-
-    let default_context = template_context! {
-        current_handle => current_handle.clone(),
-        language => language.to_string(),
-    };
-
-    let error_template = select_template!(false, true, language);
-
-    // Validate service is not empty
-    if webhook_form.service.is_empty() {
-        return contextual_error!(
-            web_context,
-            language,
-            error_template,
-            default_context,
-            "error-smokesignal-settings-3 Service cannot be empty"
-        );
-    }
-
-    // Check if service contains required the suffix
-    if !webhook_form
-        .service
-        .ends_with(SMOKE_SIGNAL_AUTOMATION_SERVICE)
-    {
-        return contextual_error!(
-            web_context,
-            language,
-            error_template,
-            default_context,
-            "error-smokesignal-settings-4 Only SmokeSignalAutomation services are supported"
-        );
-    }
-
-    // Extract DID by removing the suffix
-    let service_did = webhook_form
-        .service
-        .strip_suffix(SMOKE_SIGNAL_AUTOMATION_SERVICE)
-        .unwrap();
-
-    // Resolve the service DID using the identity resolver
-    let document = match identity_resolver.resolve(service_did).await {
-        Ok(doc) => doc,
-        Err(err) => {
-            tracing::error!(?err, "Failed to resolve service DID: {}", service_did);
-            return contextual_error!(
-                web_context,
-                language,
-                error_template,
-                default_context,
-                format!("error-smokesignal-webhook-1 DID resolution failed: {}", err)
-            );
-        }
-    };
-
-    // Store the resolved document
-    if let Err(err) = web_context
-        .document_storage
-        .store_document(document.clone())
-        .await
-    {
-        tracing::error!(?err, "Failed to store DID document for: {}", service_did);
-        return contextual_error!(
-            web_context,
-            language,
-            error_template,
-            default_context,
-            format!(
-                "error-smokesignal-webhook-2 Document storage failed: {}",
-                err
-            )
-        );
-    }
-
-    if let Err(err) = webhook_upsert(
-        &web_context.pool,
-        &current_handle.did,
-        &webhook_form.service,
-    )
-    .await
-    {
-        tracing::error!(?err, "error inserting webhook?");
-        return contextual_error!(web_context, language, error_template, default_context, err);
-    }
-    tracing::info!("webhook added?");
-
-    // Trigger HTMX refresh
-    match handle_list_webhooks(State(web_context), Language(language), Cached(auth)).await {
-        Ok(response) => Ok(response.into_response()),
-        Err(err) => Err(err),
-    }
-}
-
-pub(crate) async fn handle_toggle_webhook(
-    State(web_context): State<WebContext>,
-    Language(language): Language,
-    Cached(auth): Cached<Auth>,
-    Form(webhook_form): Form<WebhookForm>,
-) -> Result<impl IntoResponse, WebError> {
-    // Check if webhooks are enabled
-    if !web_context.config.enable_webhooks {
-        return Ok((StatusCode::NOT_FOUND, "Not found").into_response());
-    }
-
-    let current_handle = auth.require_flat()?;
-
-    let default_context = template_context! {
-        current_handle => current_handle.clone(),
-        language => language.to_string(),
-    };
-
-    let error_template = select_template!(false, true, language);
-
-    // Toggle webhook in database7
-    if let Err(err) = webhook_toggle_enabled(
-        &web_context.pool,
-        &current_handle.did,
-        &webhook_form.service,
-    )
-    .await
-    {
-        return contextual_error!(web_context, language, error_template, default_context, err);
-    }
-
-    // Trigger HTMX refresh
-    match handle_list_webhooks(State(web_context), Language(language), Cached(auth)).await {
-        Ok(response) => Ok(response.into_response()),
-        Err(err) => Err(err),
-    }
-}
-
-pub(crate) async fn handle_test_webhook(
-    State(web_context): State<WebContext>,
-    Language(language): Language,
-    Cached(auth): Cached<Auth>,
-    Form(webhook_form): Form<WebhookForm>,
-) -> Result<impl IntoResponse, WebError> {
-    // Check if webhooks are enabled
-    if !web_context.config.enable_webhooks {
-        return Ok((StatusCode::NOT_FOUND, "Not found").into_response());
-    }
-
-    let current_handle = auth.require_flat()?;
-
-    let default_context = template_context! {
-        current_handle => current_handle.clone(),
-        language => language.to_string(),
-    };
-
-    let error_template = select_template!(false, true, language);
-
-    // Send test webhook
-    if let Some(webhook_sender) = &web_context.webhook_sender {
-        if let Err(err) = webhook_sender
-            .send(TaskWork::Test {
-                identity: current_handle.did.clone(),
-                service: webhook_form.service.clone(),
-            })
-            .await
-        {
-            return contextual_error!(
-                web_context,
-                language,
-                error_template,
-                default_context,
-                format!("Failed to send webhook: {}", err)
-            );
-        }
-    } else {
-        return contextual_error!(
-            web_context,
-            language,
-            error_template,
-            default_context,
-            "Webhook processing is not enabled"
-        );
-    }
-
-    // Trigger HTMX refresh
-    match handle_list_webhooks(State(web_context), Language(language), Cached(auth)).await {
-        Ok(response) => Ok(response.into_response()),
-        Err(err) => Err(err),
-    }
-}
-
-pub(crate) async fn handle_remove_webhook(
-    State(web_context): State<WebContext>,
-    Language(language): Language,
-    Cached(auth): Cached<Auth>,
-    Form(webhook_form): Form<WebhookForm>,
-) -> Result<impl IntoResponse, WebError> {
-    // Check if webhooks are enabled
-    if !web_context.config.enable_webhooks {
-        return Ok((StatusCode::NOT_FOUND, "Not found").into_response());
-    }
-
-    let current_handle = auth.require_flat()?;
-
-    let default_context = template_context! {
-        current_handle => current_handle.clone(),
-        language => language.to_string(),
-    };
-
-    let error_template = select_template!(false, true, language);
-
-    // Remove webhook from database
-    if let Err(err) = webhook_delete(
-        &web_context.pool,
-        &current_handle.did,
-        &webhook_form.service,
-    )
-    .await
-    {
-        return contextual_error!(web_context, language, error_template, default_context, err);
-    }
-
-    // Trigger HTMX refresh
-    match handle_list_webhooks(State(web_context), Language(language), Cached(auth)).await {
-        Ok(response) => Ok(response.into_response()),
-        Err(err) => Err(err),
-    }
-}
-
-pub(crate) async fn handle_list_webhooks(
-    State(web_context): State<WebContext>,
-    Language(language): Language,
-    Cached(auth): Cached<Auth>,
-) -> Result<impl IntoResponse, WebError> {
-    // Check if webhooks are enabled
-    if !web_context.config.enable_webhooks {
-        return Ok((StatusCode::NOT_FOUND, "Not found").into_response());
-    }
-
-    let current_handle = auth.require("/settings")?;
-
-    let render_template = format!(
-        "{}/settings.webhooks.html",
-        language.to_string().to_lowercase()
-    );
-
-    // Get webhooks
-    let webhooks = webhook_list_by_did(&web_context.pool, &current_handle.did)
-        .await
-        .unwrap_or_default();
-
-    Ok((
-        StatusCode::OK,
-        RenderHtml(
-            &render_template,
-            web_context.engine.clone(),
-            template_context! {
-                current_handle => current_handle.clone(),
-                language => language.to_string(),
-                webhooks => webhooks,
-                webhooks_enabled => web_context.config.enable_webhooks,
             },
         ),
     )

src/http/server.rs

···
     handle_search::handle_search,
     handle_set_language::handle_set_language,
     handle_settings::{
-        handle_add_webhook, handle_email_update, handle_language_update, handle_list_webhooks,
+        handle_email_update, handle_language_update,
         handle_notification_email_update, handle_notification_preferences_update,
-        handle_profile_update, handle_remove_webhook, handle_settings, handle_test_webhook,
-        handle_timezone_update, handle_toggle_webhook,
+        handle_profile_update, handle_settings,
+        handle_timezone_update,
     },
     handle_unaccept_rsvp::handle_unaccept_rsvp,
     handle_unsubscribe::handle_unsubscribe,
···
             post(upload_profile_banner).layer(DefaultBodyLimit::max(5 * 1024 * 1024)), // 5MB limit
         )
         .route("/settings/banner/delete", post(delete_profile_banner))
-        .route("/settings/webhooks", get(handle_list_webhooks))
-        .route("/settings/webhooks/add", post(handle_add_webhook))
-        .route("/settings/webhooks/toggle", post(handle_toggle_webhook))
-        .route("/settings/webhooks/test", post(handle_test_webhook))
-        .route("/settings/webhooks/remove", post(handle_remove_webhook))
         .route(
             "/settings/notifications/email",
             post(handle_notification_email_update),

src/lib.rs

···
 pub mod atproto;
 pub mod config;
 pub mod config_errors;
-pub mod consumer;
 pub mod email_confirmation;
 pub mod email_errors;
 pub mod email_templates;
···
 pub mod service;
 pub mod stats;
 pub mod storage;
+pub mod tap_processor;
 pub mod task_identity_refresh;
 pub mod task_oauth_requests_cleanup;
 pub mod task_search_indexer;
 pub mod task_search_indexer_errors;
-pub mod task_webhooks;
-pub mod task_webhooks_errors;
 pub mod throttle;
 pub mod throttle_redis;
 pub mod unsubscribe_token;
 pub mod unsubscribe_token_errors;
-pub mod webhooks;

src/processor.rs

···
 use atproto_identity::model::Document;
 use atproto_identity::resolve::IdentityResolver;
 use atproto_identity::traits::DidDocumentStorage;
+use futures::future::join_all;
 use image::GenericImageView;
 use image::ImageFormat;
 use serde_json::Value;
···
 use crate::atproto::lexicon::acceptance::NSID as AcceptanceNSID;
 use crate::atproto::lexicon::acceptance::TypedAcceptance;
 use crate::atproto::lexicon::profile::{NSID as ProfileNSID, Profile};
-use crate::consumer::SmokeSignalEvent;
-use crate::consumer::SmokeSignalEventReceiver;
 use crate::processor_errors::ProcessorError;
 use crate::storage::StoragePool;
 use crate::storage::acceptance::{
···
         Ok(document)
     }
 
-    pub async fn start_processing(
+    /// Handle a commit event (create or update).
+    ///
+    /// The `live` flag indicates whether this is a live event (true) or backfill (false).
+    /// Takes ownership of `record` to avoid cloning during deserialization.
+    pub async fn handle_commit(
         &self,
-        mut event_receiver: SmokeSignalEventReceiver,
+        did: &str,
+        collection: &str,
+        rkey: &str,
+        cid: &str,
+        record: Value,
+        live: bool,
     ) -> Result<()> {
-        tracing::info!("content fetcher started");
-
-        while let Some(event) = event_receiver.recv().await {
-            match &event {
-                SmokeSignalEvent::Commit {
-                    did,
-                    collection,
-                    cid,
-                    record,
-                    rkey,
-                    ..
-                } => {
-                    let result = match collection.as_str() {
-                        "community.lexicon.calendar.event" => {
-                            self.handle_event_commit(did, rkey, cid, record).await
-                        }
-                        "community.lexicon.calendar.rsvp" => {
-                            self.handle_rsvp_commit(did, rkey, cid, record).await
-                        }
-                        "events.smokesignal.profile" => {
-                            self.handle_profile_commit(did, rkey, cid, record).await
-                        }
-                        "events.smokesignal.calendar.acceptance" => {
-                            self.handle_acceptance_commit(did, rkey, cid, record).await
-                        }
-                        _ => Ok(()),
-                    };
-                    if let Err(e) = result {
-                        tracing::error!(error = ?e, "error handling commit");
-                    }
-                }
-                SmokeSignalEvent::Delete {
-                    did,
-                    collection,
-                    rkey,
-                    ..
-                } => {
-                    let result = match collection.as_str() {
-                        "community.lexicon.calendar.event" => {
-                            self.handle_event_delete(did, rkey).await
-                        }
-                        "community.lexicon.calendar.rsvp" => {
-                            self.handle_rsvp_delete(did, rkey).await
-                        }
-                        "events.smokesignal.profile" => self.handle_profile_delete(did, rkey).await,
-                        "events.smokesignal.calendar.acceptance" => {
-                            self.handle_acceptance_delete(did, rkey).await
-                        }
-                        _ => Ok(()),
-                    };
-                    if let Err(e) = result {
-                        tracing::error!(error = ?e, "error handling delete");
-                    }
-                }
+        match collection {
+            "community.lexicon.calendar.event" => {
+                self.handle_event_commit(did, rkey, cid, record, live)
+                    .await
+            }
+            "community.lexicon.calendar.rsvp" => {
+                self.handle_rsvp_commit(did, rkey, cid, record, live).await
+            }
+            "events.smokesignal.profile" => {
+                self.handle_profile_commit(did, rkey, cid, record).await
+            }
+            "events.smokesignal.calendar.acceptance" => {
+                self.handle_acceptance_commit(did, rkey, cid, record).await
             }
+            _ => Ok(()),
         }
+    }
 
-        tracing::info!("content fetcher finished");
-        Ok(())
+    /// Handle a delete event.
+    ///
+    /// The `live` flag indicates whether this is a live event (true) or backfill (false).
+    pub async fn handle_delete(
+        &self,
+        did: &str,
+        collection: &str,
+        rkey: &str,
+        live: bool,
+    ) -> Result<()> {
+        match collection {
+            "community.lexicon.calendar.event" => {
+                self.handle_event_delete(did, rkey, live).await
+            }
+            "community.lexicon.calendar.rsvp" => self.handle_rsvp_delete(did, rkey, live).await,
+            "events.smokesignal.profile" => self.handle_profile_delete(did, rkey).await,
+            "events.smokesignal.calendar.acceptance" => {
+                self.handle_acceptance_delete(did, rkey).await
+            }
+            _ => Ok(()),
+        }
     }
 
     async fn handle_event_commit(
···
         did: &str,
         rkey: &str,
         cid: &str,
-        record: &Value,
+        record: Value,
+        _live: bool,
     ) -> Result<()> {
         tracing::info!("Processing event: {} for {}", rkey, did);
 
         let aturi = format!("at://{did}/{LexiconCommunityEventNSID}/{rkey}");
 
-        let event_record: Event = serde_json::from_value(record.clone())?;
+        let event_record: Event = serde_json::from_value(record)?;
 
         let document = self.ensure_identity_stored(did).await?;
         let pds_endpoints = document.pds_endpoints();
···
 
         let all_media = event_record.media;
 
-        for media in &all_media {
-            if let Err(err) = self.download_media(pds_endpoint, did, media).await {
-                tracing::error!(error = ?err, "failed processing image");
+        // Download all media items in parallel
+        if !all_media.is_empty() {
+            let download_futures: Vec<_> = all_media
+                .iter()
+                .map(|media| self.download_media(pds_endpoint, did, media))
+                .collect();
+
+            let results = join_all(download_futures).await;
+
+            for result in results {
+                if let Err(err) = result {
+                    tracing::error!(error = ?err, "failed processing image");
+                }
             }
         }
 
···
         did: &str,
         rkey: &str,
         cid: &str,
-        record: &Value,
+        record: Value,
+        _live: bool,
     ) -> Result<()> {
         tracing::info!("Processing rsvp: {} for {}", rkey, did);
 
         let aturi = format!("at://{did}/{LexiconCommunityRSVPNSID}/{rkey}");
 
-        let rsvp_record: Rsvp = serde_json::from_value(record.clone())?;
+        let rsvp_record: Rsvp = serde_json::from_value(record)?;
 
         let event_aturi = rsvp_record.subject.uri.clone();
         let event_cid = rsvp_record.subject.cid.clone();
···
         Ok(())
     }
 
-    async fn handle_event_delete(&self, did: &str, rkey: &str) -> Result<()> {
+    async fn handle_event_delete(&self, did: &str, rkey: &str, _live: bool) -> Result<()> {
         let aturi = format!("at://{did}/{LexiconCommunityEventNSID}/{rkey}");
 
         event_delete(&self.pool, &aturi).await?;
···
         Ok(())
     }
 
-    async fn handle_rsvp_delete(&self, did: &str, rkey: &str) -> Result<()> {
+    async fn handle_rsvp_delete(&self, did: &str, rkey: &str, _live: bool) -> Result<()> {
         let aturi = format!("at://{did}/{LexiconCommunityEventNSID}/{rkey}");
 
         rsvp_delete(&self.pool, &aturi).await?;
···
         did: &str,
         rkey: &str,
         cid: &str,
-        record: &Value,
+        record: Value,
     ) -> Result<()> {
         tracing::info!("Processing profile: {} for {}", rkey, did);
 
···
             return Ok(());
         }
 
-        let profile_record: Profile = serde_json::from_value(record.clone())?;
+        let profile_record: Profile = serde_json::from_value(record)?;
 
         // Get the identity to resolve the handle for display_name fallback and PDS endpoint
         let document = self.ensure_identity_stored(did).await?;
···
 
         profile_insert(&self.pool, &aturi, cid, did, display_name, &profile_record).await?;
 
-        // Download avatar and banner blobs if present
+        // Download avatar and banner blobs if present (in parallel)
         let pds_endpoints = document.pds_endpoints();
         if let Some(pds_endpoint) = pds_endpoints.first() {
-            // Download avatar if present
-            if let Some(ref avatar) = profile_record.avatar
-                && let Err(e) = self.download_avatar(pds_endpoint, did, avatar).await
-            {
+            // Create futures for avatar and banner downloads
+            let avatar_future = async {
+                if let Some(ref avatar) = profile_record.avatar {
+                    self.download_avatar(pds_endpoint, did, avatar).await
+                } else {
+                    Ok(())
+                }
+            };
+
+            let banner_future = async {
+                if let Some(ref banner) = profile_record.banner {
+                    self.download_banner(pds_endpoint, did, banner).await
+                } else {
+                    Ok(())
+                }
+            };
+
+            // Download both concurrently
+            let (avatar_result, banner_result) = tokio::join!(avatar_future, banner_future);
+
+            if let Err(e) = avatar_result {
                 tracing::warn!(
                     error = ?e,
                     did = %did,
···
                 );
             }
 
-            // Download banner if present
-            if let Some(ref banner) = profile_record.banner
-                && let Err(e) = self.download_banner(pds_endpoint, did, banner).await
-            {
+            if let Err(e) = banner_result {
                 tracing::warn!(
                     error = ?e,
                     did = %did,
···
         did: &str,
         rkey: &str,
         cid: &str,
-        record: &Value,
+        record: Value,
     ) -> Result<()> {
         tracing::info!("Processing acceptance: {} for {}", rkey, did);
 
         let aturi = format!("at://{did}/{AcceptanceNSID}/{rkey}");
 
         // Deserialize and validate the acceptance record
-        let acceptance_record: TypedAcceptance = serde_json::from_value(record.clone())?;
+        let acceptance_record: TypedAcceptance = serde_json::from_value(record)?;
         tracing::info!(?acceptance_record, "acceptance_record");
 
         // Validate the acceptance record
···
             return Ok(());
         }
 
-        // Store the acceptance record
-        acceptance_record_upsert(&self.pool, &aturi, cid, did, record).await?;
+        // Store the acceptance record (use deserialized record to avoid clone)
+        acceptance_record_upsert(&self.pool, &aturi, cid, did, &acceptance_record).await?;
 
         tracing::info!("Acceptance stored: {}", aturi);
         Ok(())
···
             img
         };
 
-        // Convert to PNG
-        let mut png_buffer = std::io::Cursor::new(Vec::new());
+        // Convert to PNG with pre-allocated buffer
+        // 400x400 PNG typically compresses to ~100-300KB, pre-allocate 256KB
+        let mut png_buffer = std::io::Cursor::new(Vec::with_capacity(256 * 1024));
         resized.write_to(&mut png_buffer, ImageFormat::Png)?;
         let png_bytes = png_buffer.into_inner();
 
···
             img
         };
 
-        // Convert to PNG
-        let mut png_buffer = std::io::Cursor::new(Vec::new());
+        // Convert to PNG with pre-allocated buffer
+        // 1600x900 PNG typically compresses to ~500KB-1.5MB, pre-allocate 1MB
+        let mut png_buffer = std::io::Cursor::new(Vec::with_capacity(1024 * 1024));
         resized.write_to(&mut png_buffer, ImageFormat::Png)?;
         let png_bytes = png_buffer.into_inner();
 
···
             img
         };
 
-        let mut png_buffer = std::io::Cursor::new(Vec::new());
+        // Convert to PNG with pre-allocated buffer
+        // 1600x900 PNG typically compresses to ~500KB-1.5MB, pre-allocate 1MB
+        let mut png_buffer = std::io::Cursor::new(Vec::with_capacity(1024 * 1024));
         final_image.write_to(&mut png_buffer, ImageFormat::Png)?;
         let png_bytes = png_buffer.into_inner();
 

src/storage/mod.rs

···
 pub mod private_event_content;
 pub mod profile;
 pub mod types;
-pub mod webhook;
 pub use types::*;

src/storage/webhook.rs

···
-use chrono::Utc;
-
-use crate::storage::{StoragePool, errors::StorageError};
-
-pub(crate) mod model {
-    use chrono::{DateTime, Utc};
-    use serde::{Deserialize, Serialize};
-    use sqlx::FromRow;
-
-    #[derive(Clone, FromRow, Deserialize, Serialize, Debug)]
-    pub struct IdentityWebhook {
-        pub did: String,
-        pub service: String,
-        pub created_at: DateTime<Utc>,
-        pub enabled: bool,
-        pub errored_at: Option<DateTime<Utc>>,
-        pub error: Option<String>,
-    }
-}
-
-pub use model::IdentityWebhook;
-
-pub async fn webhook_upsert(
-    pool: &StoragePool,
-    did: &str,
-    service: &str,
-) -> Result<(), StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    let now = Utc::now();
-
-    sqlx::query(
-        r#"
-        INSERT INTO identity_webhooks (did, service, created_at, enabled, errored_at, error)
-        VALUES ($1, $2, $3, true, NULL, NULL)
-        ON CONFLICT (did, service) DO UPDATE SET
-            enabled = true,
-            errored_at = NULL,
-            error = NULL
-        "#,
-    )
-    .bind(did)
-    .bind(service)
-    .bind(now)
-    .execute(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(())
-}
-
-pub async fn webhook_toggle_enabled(
-    pool: &StoragePool,
-    did: &str,
-    service: &str,
-) -> Result<(), StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    sqlx::query(
-        r#"
-        UPDATE identity_webhooks
-        SET enabled = NOT enabled
-        WHERE did = $1 AND service = $2
-        "#,
-    )
-    .bind(did)
-    .bind(service)
-    .execute(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(())
-}
-
-pub async fn webhook_list_by_did(
-    pool: &StoragePool,
-    did: &str,
-) -> Result<Vec<IdentityWebhook>, StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    let webhooks = sqlx::query_as::<_, model::IdentityWebhook>(
-        r#"
-        SELECT did, service, created_at, enabled, errored_at, error
-        FROM identity_webhooks
-        WHERE did = $1
-        ORDER BY created_at DESC
-        "#,
-    )
-    .bind(did)
-    .fetch_all(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(webhooks)
-}
-
-pub async fn webhook_delete(
-    pool: &StoragePool,
-    did: &str,
-    service: &str,
-) -> Result<(), StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    sqlx::query(
-        r#"
-        DELETE FROM identity_webhooks
-        WHERE did = $1 AND service = $2
-        "#,
-    )
-    .bind(did)
-    .bind(service)
-    .execute(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(())
-}
-
-pub async fn webhook_failed(
-    pool: &StoragePool,
-    did: &str,
-    service: &str,
-    error: &str,
-) -> Result<(), StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    let now = Utc::now();
-
-    sqlx::query(
-        r#"
-        UPDATE identity_webhooks
-        SET enabled = false,
-            errored_at = $4,
-            error = $3
-        WHERE did = $1 AND service = $2
-        "#,
-    )
-    .bind(did)
-    .bind(service)
-    .bind(error)
-    .bind(now)
-    .execute(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(())
-}
-
-pub async fn webhook_list_enabled_by_did(
-    pool: &StoragePool,
-    did: &str,
-) -> Result<Vec<IdentityWebhook>, StorageError> {
-    let mut tx = pool
-        .begin()
-        .await
-        .map_err(StorageError::CannotBeginDatabaseTransaction)?;
-
-    let webhooks = sqlx::query_as::<_, model::IdentityWebhook>(
-        r#"
-        SELECT did, service, created_at, enabled, errored_at, error
-        FROM identity_webhooks
-        WHERE did = $1 AND enabled = true
-        ORDER BY created_at DESC
-        "#,
-    )
-    .bind(did)
-    .fetch_all(tx.as_mut())
-    .await
-    .map_err(StorageError::UnableToExecuteQuery)?;
-
-    tx.commit()
-        .await
-        .map_err(StorageError::CannotCommitDatabaseTransaction)?;
-
-    Ok(webhooks)
-}

src/tap_processor.rs

···
+//! TAP event processor for AT Protocol events.
+//!
+//! This module provides a unified TAP stream consumer and event processor,
+//! eliminating the need for channel-based message passing.
+
+use anyhow::Result;
+use atproto_tap::{connect, RecordAction, TapConfig, TapEvent};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio_stream::StreamExt;
+use tokio_util::sync::CancellationToken;
+
+use crate::processor::ContentFetcher;
+use crate::task_search_indexer::SearchIndexer;
+
+/// Unified TAP stream consumer and event processor.
+///
+/// Combines TAP event consumption with direct processing, eliminating
+/// channel overhead. Events are processed inline as they arrive from
+/// the TAP stream.
+pub struct TapProcessor {
+    config: TapConfig,
+    content_fetcher: ContentFetcher,
+    search_indexer: Option<SearchIndexer>,
+    cancel_token: CancellationToken,
+}
+
+impl TapProcessor {
+    /// Create a new TAP processor.
+    ///
+    /// # Arguments
+    ///
+    /// * `hostname` - TAP service hostname (e.g., "localhost:2480")
+    /// * `password` - Optional admin password for authentication
+    /// * `user_agent` - User-Agent header for WebSocket connections
+    /// * `content_fetcher` - Processor for event/RSVP/profile content
+    /// * `search_indexer` - Optional search indexer for OpenSearch
+    /// * `cancel_token` - Token for graceful shutdown
+    pub fn new(
+        hostname: &str,
+        password: Option<String>,
+        user_agent: &str,
+        content_fetcher: ContentFetcher,
+        search_indexer: Option<SearchIndexer>,
+        cancel_token: CancellationToken,
+    ) -> Self {
+        let mut config_builder = TapConfig::builder()
+            .hostname(hostname)
+            .send_acks(true)
+            .user_agent(user_agent)
+            .initial_reconnect_delay(Duration::from_secs(1))
+            .max_reconnect_delay(Duration::from_secs(30));
+
+        if let Some(password) = password {
+            config_builder = config_builder.admin_password(password);
+        }
+
+        Self {
+            config: config_builder.build(),
+            content_fetcher,
+            search_indexer,
+            cancel_token,
+        }
+    }
+
+    /// Run the TAP processor, consuming events until cancelled.
+    pub async fn run(self) -> Result<()> {
+        tracing::info!(
+            "TAP processor starting, connecting to {}",
+            self.config.hostname
+        );
+        let mut stream = connect(self.config.clone());
+
+        loop {
+            tokio::select! {
+                () = self.cancel_token.cancelled() => {
+                    tracing::info!("TAP processor cancelled, closing stream");
+                    stream.close().await;
+                    break;
+                }
+                Some(result) = stream.next() => {
+                    match result {
+                        Ok(event) => {
+                            self.process_event(&event).await;
+                        }
+                        Err(err) => {
+                            if err.is_fatal() {
+                                tracing::error!(?err, "Fatal TAP error, exiting processor");
+                                break;
+                            }
+                            tracing::warn!(?err, "TAP stream error (will retry)");
+                        }
+                    }
+                }
+            }
+        }
+
+        tracing::info!("TAP processor stopped");
+        Ok(())
+    }
+
+    async fn process_event(&self, event: &Arc<TapEvent>) {
+        match event.as_ref() {
+            TapEvent::Record { record, .. } => {
+                let collection = record.collection.as_ref();
+
+                // Skip irrelevant collections (TAP should filter, but double-check)
+                if !Self::is_relevant_collection(collection) {
+                    return;
+                }
+
+                let did = record.did.as_ref();
+                let rkey = record.rkey.as_ref();
+                let live = record.live;
+
+                if !live {
+                    tracing::debug!(
+                        "Processing backfill event: {} {} {}",
+                        collection,
+                        did,
+                        rkey
+                    );
+                }
+
+                match record.action {
+                    RecordAction::Create | RecordAction::Update => {
+                        let cid = record.cid.as_ref().map(|c| c.as_ref()).unwrap_or("");
+                        let record_value = record
+                            .record
+                            .as_ref()
+                            .cloned()
+                            .unwrap_or(serde_json::Value::Null);
+
+                        // Process content fetcher and search indexer in parallel
+                        // Clone record_value for search indexer so both can take ownership
+                        let indexer_record = if self.search_indexer.is_some() {
+                            Some(record_value.clone())
+                        } else {
+                            None
+                        };
+
+                        let content_future = self
+                            .content_fetcher
+                            .handle_commit(did, collection, rkey, cid, record_value, live);
+
+                        let index_future = async {
+                            if let (Some(indexer), Some(record)) =
+                                (&self.search_indexer, indexer_record)
+                            {
+                                indexer.index_commit(did, collection, rkey, record).await
+                            } else {
+                                Ok(())
+                            }
+                        };
+
+                        let (content_result, index_result) =
+                            tokio::join!(content_future, index_future);
+
+                        if let Err(e) = content_result {
+                            tracing::error!(?e, "Error processing commit");
+                        }
+                        if let Err(e) = index_result {
+                            tracing::error!(?e, "Error indexing commit");
+                        }
+                    }
+                    RecordAction::Delete => {
+                        // Process content fetcher and search indexer delete in parallel
+                        let content_future =
+                            self.content_fetcher
+                                .handle_delete(did, collection, rkey, live);
+
+                        let index_future = async {
+                            if let Some(ref indexer) = self.search_indexer {
+                                indexer.delete_record(did, collection, rkey).await
+                            } else {
+                                Ok(())
+                            }
+                        };
+
+                        let (content_result, index_result) =
+                            tokio::join!(content_future, index_future);
+
+                        if let Err(e) = content_result {
+                            tracing::error!(?e, "Error processing delete");
+                        }
+                        if let Err(e) = index_result {
+                            tracing::error!(?e, "Error deleting from index");
+                        }
+                    }
+                }
+            }
+            TapEvent::Identity { identity, .. } => {
+                // Identity events ignored (handle updates can be added in future iteration)
+                tracing::trace!(
+                    did = %identity.did,
+                    handle = %identity.handle,
+                    "Identity event ignored"
+                );
+            }
+        }
+    }
+
+    fn is_relevant_collection(collection: &str) -> bool {
+        matches!(
+            collection,
+            "community.lexicon.calendar.rsvp"
+                | "community.lexicon.calendar.event"
+                | "events.smokesignal.profile"
+                | "events.smokesignal.calendar.acceptance"
+        )
+    }
+}

src/task_search_indexer.rs

···
 use serde::Deserialize;
 use serde_json::{Value, json};
 use std::sync::Arc;
-use tokio_util::sync::CancellationToken;
 
 use crate::atproto::lexicon::profile::{Profile, NSID as PROFILE_NSID};
 use crate::atproto::utils::get_profile_hashtags;
-use crate::consumer::{SmokeSignalEvent, SmokeSignalEventReceiver};
 use crate::task_search_indexer_errors::SearchIndexerError;
 
 /// A lightweight event struct for search indexing that excludes problematic fields like locations.
···
     client: Arc<OpenSearch>,
     identity_resolver: Arc<dyn IdentityResolver>,
     document_storage: Arc<dyn DidDocumentStorage + Send + Sync>,
-    receiver: SmokeSignalEventReceiver,
-    cancel_token: CancellationToken,
 }
 
 impl SearchIndexer {
+    /// Create a new SearchIndexer.
+    ///
+    /// # Arguments
+    ///
+    /// * `endpoint` - OpenSearch endpoint URL
+    /// * `identity_resolver` - Resolver for DID identities
+    /// * `document_storage` - Storage for DID documents
     pub async fn new(
         endpoint: &str,
         identity_resolver: Arc<dyn IdentityResolver>,
         document_storage: Arc<dyn DidDocumentStorage + Send + Sync>,
-        receiver: SmokeSignalEventReceiver,
-        cancel_token: CancellationToken,
     ) -> Result<Self> {
         let transport = Transport::single_node(endpoint)?;
         let client = Arc::new(OpenSearch::new(transport));
···
             client,
             identity_resolver,
             document_storage,
-            receiver,
-            cancel_token,
         };
 
         indexer.ensure_index().await?;
···
         Ok(())
     }
 
-    pub async fn run(mut self) -> Result<()> {
-        tracing::info!("Search indexer task started");
-
-        loop {
-            tokio::select! {
-                _ = self.cancel_token.cancelled() => {
-                    tracing::info!("Search indexer task shutting down");
-                    break;
-                }
-                Some(event) = self.receiver.recv() => {
-                    if let Err(err) = self.process_event(event).await {
-                        tracing::error!(?err, "Error processing event for search index");
-                    }
-                }
-            }
+    /// Index a commit event (create or update).
+    ///
+    /// Dispatches to the appropriate indexer based on collection type.
+    /// Takes ownership of `record` to avoid cloning during deserialization.
+    pub async fn index_commit(
+        &self,
+        did: &str,
+        collection: &str,
+        rkey: &str,
+        record: Value,
+    ) -> Result<()> {
+        match collection {
+            "community.lexicon.calendar.event" => self.index_event(did, rkey, record).await,
+            c if c == PROFILE_NSID => self.index_profile(did, rkey, record).await,
+            _ => Ok(()),
         }
-
-        Ok(())
     }
 
-    async fn process_event(&self, event: SmokeSignalEvent) -> Result<()> {
-        match event {
-            SmokeSignalEvent::Commit {
-                did,
-                collection,
-                rkey,
-                record,
-                ..
-            } => {
-                if collection == "community.lexicon.calendar.event" {
-                    self.index_event(&did, &rkey, &record).await?;
-                } else if collection == PROFILE_NSID {
-                    self.index_profile(&did, &rkey, &record).await?;
-                }
-            }
-            SmokeSignalEvent::Delete {
-                did,
-                collection,
-                rkey,
-                ..
-            } => {
-                if collection == "community.lexicon.calendar.event" {
-                    self.delete_event(&did, &rkey).await?;
-                } else if collection == PROFILE_NSID {
-                    self.delete_profile(&did, &rkey).await?;
-                }
-            }
+    /// Delete a record from the search index.
+    ///
+    /// Dispatches to the appropriate delete method based on collection type.
+    pub async fn delete_record(&self, did: &str, collection: &str, rkey: &str) -> Result<()> {
+        match collection {
+            "community.lexicon.calendar.event" => self.delete_event(did, rkey).await,
+            c if c == PROFILE_NSID => self.delete_profile(did, rkey).await,
+            _ => Ok(()),
         }
-
-        Ok(())
     }
 
-    async fn index_event(&self, did: &str, rkey: &str, record: &Value) -> Result<()> {
+    async fn index_event(&self, did: &str, rkey: &str, record: Value) -> Result<()> {
         // Use IndexableEvent which excludes problematic fields like locations
-        let event: IndexableEvent = serde_json::from_value(record.clone())?;
+        let event: IndexableEvent = serde_json::from_value(record)?;
 
         let document = self.ensure_identity_stored(did).await?;
         let handle = document.handles().unwrap_or("invalid.handle");
···
         Ok(())
     }
 
-    async fn index_profile(&self, did: &str, rkey: &str, record: &Value) -> Result<()> {
-        let profile: Profile = serde_json::from_value(record.clone())?;
+    async fn index_profile(&self, did: &str, rkey: &str, record: Value) -> Result<()> {
+        let profile: Profile = serde_json::from_value(record)?;
 
         let document = self.ensure_identity_stored(did).await?;
         let handle = document.handles().unwrap_or("invalid.handle");

src/task_webhooks.rs

···
-use anyhow::Result;
-use atproto_identity::traits::DidDocumentStorage;
-use atproto_oauth::jwt::{Claims, Header, mint};
-use chrono::Utc;
-use rand::distr::{Alphanumeric, SampleString};
-use serde::{Deserialize, Serialize};
-use serde_json::json;
-use sqlx::PgPool;
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::sync::mpsc;
-use tokio_util::sync::CancellationToken;
-
-use crate::{
-    service::{ServiceDID, ServiceKey},
-    storage::webhook::webhook_failed,
-    task_webhooks_errors::WebhookError,
-    webhooks::{
-        EVENT_CREATED_EVENT, RSVP_CREATED_EVENT, SMOKE_SIGNAL_AUTOMATION_SERVICE, TEST_EVENT,
-    },
-};
-
-#[derive(Debug, Clone)]
-pub enum TaskWork {
-    Test {
-        identity: String,
-        service: String,
-    },
-    RSVPCreated {
-        identity: String,
-        service: String,
-        record: serde_json::Value,
-        context: serde_json::Value,
-    },
-    EventCreated {
-        identity: String,
-        service: String,
-        record: serde_json::Value,
-        context: serde_json::Value,
-    },
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-struct WebhookPayload {
-    record: serde_json::Value,
-    context: serde_json::Value,
-    event: String,
-}
-
-pub struct WebhookProcessor {
-    pool: PgPool,
-    document_storage: Arc<dyn DidDocumentStorage>,
-    service_did: ServiceDID,
-    service_key: ServiceKey,
-    receiver: mpsc::Receiver<TaskWork>,
-    cancel_token: CancellationToken,
-}
-
-impl WebhookProcessor {
-    pub fn new(
-        pool: PgPool,
-        document_storage: Arc<dyn DidDocumentStorage>,
-        service_did: ServiceDID,
-        service_key: ServiceKey,
-        receiver: mpsc::Receiver<TaskWork>,
-        cancel_token: CancellationToken,
-    ) -> Self {
-        Self {
-            pool,
-            document_storage,
-            service_did,
-            service_key,
-            receiver,
-            cancel_token,
-        }
-    }
-
-    pub async fn run(mut self) -> Result<()> {
-        tracing::info!("Webhook processor task started");
-
-        loop {
-            tokio::select! {
-                _ = self.cancel_token.cancelled() => {
-                    tracing::info!("Webhook processor task shutting down");
-                    break;
-                }
-                Some(work) = self.receiver.recv() => {
-                    if let Err(e) = self.process_work(work).await {
-                        tracing::error!("Error processing webhook work: {:?}", e);
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn process_work(&self, work: TaskWork) -> Result<()> {
-        match work {
-            TaskWork::Test { identity, service } => {
-                self.send_webhook(
-                    &identity,
-                    &service,
-                    WebhookPayload {
-                        record: json!({}),
-                        context: json!({}),
-                        event: TEST_EVENT.to_string(),
-                    },
-                )
-                .await
-            }
-            TaskWork::RSVPCreated {
-                identity,
-                service,
-                record,
-                context,
-            } => {
-                self.send_webhook(
-                    &identity,
-                    &service,
-                    WebhookPayload {
-                        record,
-                        context,
-                        event: RSVP_CREATED_EVENT.to_string(),
-                    },
-                )
-                .await
-            }
-            TaskWork::EventCreated {
-                identity,
-                service,
-                record,
-                context,
-            } => {
-                self.send_webhook(
-                    &identity,
-                    &service,
-                    WebhookPayload {
-                        record,
-                        context,
-                        event: EVENT_CREATED_EVENT.to_string(),
-                    },
-                )
-                .await
-            }
-        }
-    }
-
-    async fn send_webhook(
-        &self,
-        identity: &str,
-        service: &str,
-        payload: WebhookPayload,
-    ) -> Result<()> {
-        // Remove the suffix from service
-        if !service.ends_with(SMOKE_SIGNAL_AUTOMATION_SERVICE) {
-            return Err(WebhookError::InvalidServiceSuffix {
-                suffix: SMOKE_SIGNAL_AUTOMATION_SERVICE.to_string(),
-            }
-            .into());
-        }
-
-        let service_did = service
-            .strip_suffix(SMOKE_SIGNAL_AUTOMATION_SERVICE)
-            .unwrap();
-
-        // Get the DID document from document storage
-        let document = self
-            .document_storage
-            .get_document_by_did(service_did)
-            .await
-            .map_err(|e| WebhookError::DidDocumentRetrievalFailed {
-                did: service_did.to_string(),
-                error: e.to_string(),
-            })?
-            .ok_or_else(|| WebhookError::DidDocumentNotFound {
-                did: service_did.to_string(),
-            })?;
-
-        // Extract the service endpoint
-        let automation_service = document
-            .service
-            .iter()
-            .find(|service| service.id.ends_with(SMOKE_SIGNAL_AUTOMATION_SERVICE))
-            .ok_or(WebhookError::ServiceNotFoundInDidDocument)?;
-
-        // Get the service endpoint - it should be a string URL
-        let endpoint_url = &automation_service.service_endpoint;
-
-        // Construct the webhook URL
-        let webhook_url = format!(
-            "{}/xrpc/events.smokesignal.automation.InvokeWebhook",
-            endpoint_url
-        );
-
-        // Create signed JWT
-        let header: Header = self.service_key.0.clone().try_into()?;
-
-        let now = Utc::now();
-        let jti = Alphanumeric.sample_string(&mut rand::rng(), 30);
-
-        // Create base claims
-        let mut claims = Claims::default();
-        claims.jose.issuer = Some(self.service_did.0.clone());
-        claims.jose.audience = Some(service_did.to_string());
-        claims.jose.issued_at = Some(now.timestamp() as u64);
-        claims.jose.expiration = Some((now.timestamp() as u64) + 60);
-        claims.jose.json_web_token_id = Some(jti);
-        claims.private.insert(
-            "lxm".to_string(),
-            serde_json::Value::String("events.smokesignal.automation.InvokeWebhook".to_string()),
-        );
-
-        let token = mint(&self.service_key.0, &header, &claims).map_err(|e| {
-            WebhookError::JwtCreationFailed {
-                error: e.to_string(),
-            }
-        })?;
-
-        // Prepare headers with JWT authorization
-        let mut headers = reqwest::header::HeaderMap::new();
-        headers.insert("authorization", format!("Bearer {}", token).parse()?);
-        headers.insert("content-type", "application/json".parse()?);
-
-        // Send the webhook
-        let client = reqwest::Client::new();
-        let response = client
-            .post(&webhook_url)
-            .headers(headers)
-            .json(&payload)
-            .timeout(Duration::from_secs(30))
-            .send()
-            .await;
-
-        match response {
-            Ok(resp) if resp.status().is_success() => {
-                tracing::info!("Webhook sent successfully to {} for {}", service, identity);
-                Ok(())
-            }
-            Ok(resp) => {
-                let status = resp.status();
-                let error_text = resp
-                    .text()
-                    .await
-                    .unwrap_or_else(|_| "Unknown error".to_string());
-                let error_msg = format!("HTTP {}: {}", status, error_text);
-
-                tracing::error!("Webhook failed: {}", error_msg);
-
-                // Update database to mark webhook as failed
-                webhook_failed(&self.pool, identity, service, &error_msg).await?;
-
-                Err(WebhookError::WebhookRequestFailed { error: error_msg }.into())
-            }
-            Err(e) => {
-                let error_msg = format!("Request failed: {}", e);
-                tracing::error!("Webhook request error: {}", error_msg);
-
-                // Update database to mark webhook as failed
-                webhook_failed(&self.pool, identity, service, &error_msg).await?;
-
-                Err(WebhookError::WebhookTransportFailed {
-                    error: e.to_string(),
-                }
-                .into())
-            }
-        }
-    }
-}

src/task_webhooks_errors.rs

···
-use thiserror::Error;
-
-/// Represents errors that can occur during webhook processing.
-///
-/// These errors typically occur when processing webhook notifications,
-/// including service validation, DID resolution, and HTTP request failures.
-#[derive(Debug, Error)]
-pub(crate) enum WebhookError {
-    /// Error when a service doesn't end with the required suffix.
-    ///
-    /// This error occurs when validating a webhook service URL that doesn't
-    /// end with the expected automation service suffix.
-    #[error("error-smokesignal-webhook-1 Service must end with {suffix}")]
-    InvalidServiceSuffix { suffix: String },
-
-    /// Error when failing to get a DID document.
-    ///
-    /// This error occurs when attempting to retrieve a DID document
-    /// for service validation fails.
-    #[error("error-smokesignal-webhook-2 Failed to get DID document for {did}: {error}")]
-    DidDocumentRetrievalFailed { did: String, error: String },
-
-    /// Error when a DID document is not found.
-    ///
-    /// This error occurs when a required DID document cannot be found
-    /// in the system.
-    #[error("error-smokesignal-webhook-3 DID document not found for {did}")]
-    DidDocumentNotFound { did: String },
-
-    /// Error when a service is not found in a DID document.
-    ///
-    /// This error occurs when a DID document doesn't contain the expected
-    /// automation service entry.
-    #[error("error-smokesignal-webhook-4 service not found in DID document")]
-    ServiceNotFoundInDidDocument,
-
-    /// Error when JWT creation fails.
-    ///
-    /// This error occurs when attempting to create a JWT token for
-    /// webhook authentication fails.
-    #[error("error-smokesignal-webhook-5 Failed to create JWT: {error}")]
-    JwtCreationFailed { error: String },
-
-    /// Error when a webhook request fails with an error response.
-    ///
-    /// This error occurs when the webhook endpoint returns an error
-    /// status code or the request fails.
-    #[error("error-smokesignal-webhook-6 Webhook failed: {error}")]
-    WebhookRequestFailed { error: String },
-
-    /// Error when a webhook request fails due to network or other issues.
-    ///
-    /// This error occurs when the HTTP request to the webhook endpoint
-    /// fails due to network issues or other transport problems.
-    #[error("error-smokesignal-webhook-7 Webhook request failed: {error}")]
-    WebhookTransportFailed { error: String },
-}

src/webhooks.rs

···
-pub(crate) const SMOKE_SIGNAL_AUTOMATION_SERVICE: &str = "#SmokeSignalAutomation";
-
-pub(crate) const RSVP_CREATED_EVENT: &str = "rsvp.created";
-pub(crate) const EVENT_CREATED_EVENT: &str = "event.created";
-pub(crate) const TEST_EVENT: &str = "test";

templates/en-us/settings.common.html

···
                     </div>
                 </div>
             </div>
-
-            {% if webhooks_enabled %}
-            <hr>
-
-            <div class="columns">
-                <div class="column">
-                    <h2 class="subtitle">Webhooks</h2>
-                    <p class="mb-4">Configure webhook endpoints to receive notifications when events occur.</p>
-
-                    <div id="webhooks-list">
-                        {% include "en-us/settings.webhooks.html" %}
-                    </div>
-
-                    <div class="box mt-4">
-                        <h3 class="subtitle is-5">Add New Webhook</h3>
-                        <form hx-post="/settings/webhooks/add" hx-target="#webhooks-list" hx-swap="innerHTML">
-                            <div class="field has-addons">
-                                <div class="control is-expanded">
-                                    <input class="input" type="text" name="service" placeholder="webhook-service-name" required>
-                                </div>
-                                <div class="control">
-                                    <button class="button is-primary" type="submit">
-                                        <span class="icon">
-                                            <i class="fas fa-plus"></i>
-                                        </span>
-                                        <span>Add Webhook</span>
-                                    </button>
-                                </div>
-                            </div>
-                            <p class="help">Enter the service name for your webhook endpoint (e.g., "my-automation-service")</p>
-                        </form>
-                    </div>
-                </div>
-            </div>
-            {% endif %}
         </div>
     </div>
 </div>

templates/en-us/settings.webhooks.html

···
-{% if webhooks %}
-<div class="table-container">
-    <table class="table is-fullwidth is-striped">
-        <thead>
-            <tr>
-                <th>Service</th>
-                <th>Status</th>
-                <th>Created</th>
-                <th>Last Error</th>
-                <th>Actions</th>
-            </tr>
-        </thead>
-        <tbody>
-            {% for webhook in webhooks %}
-            <tr>
-                <td>{{ webhook.service }}</td>
-                <td>
-                    {% if webhook.enabled %}
-                        <span class="tag is-success">Enabled</span>
-                    {% else %}
-                        <span class="tag is-danger">Disabled</span>
-                    {% endif %}
-                </td>
-                <td>{{ webhook.created_at }}</td>
-                <td>
-                    {% if webhook.error %}
-                        <span class="has-text-danger" title="{{ webhook.error }}">
-                            {{ webhook.errored_at }}
-                        </span>
-                    {% else %}
-                        <span class="has-text-grey">-</span>
-                    {% endif %}
-                </td>
-                <td>
-                    <div class="buttons are-small">
-                        <form hx-post="/settings/webhooks/toggle" hx-target="#webhooks-list" hx-swap="innerHTML" style="display: inline;">
-                            <input type="hidden" name="service" value="{{ webhook.service }}">
-                            <button class="button is-small {% if webhook.enabled %}is-warning{% else %}is-success{% endif %}" type="submit">
-                                <span class="icon">
-                                    <i class="fas {% if webhook.enabled %}fa-pause{% else %}fa-play{% endif %}"></i>
-                                </span>
-                                <span>{% if webhook.enabled %}Disable{% else %}Enable{% endif %}</span>
-                            </button>
-                        </form>
-                        
-                        <form hx-post="/settings/webhooks/test" hx-target="#webhooks-list" hx-swap="innerHTML" style="display: inline;">
-                            <input type="hidden" name="service" value="{{ webhook.service }}">
-                            <button class="button is-small is-info" type="submit" {% if not webhook.enabled %}disabled{% endif %}>
-                                <span class="icon">
-                                    <i class="fas fa-vial"></i>
-                                </span>
-                                <span>Test</span>
-                            </button>
-                        </form>
-                        
-                        <form hx-post="/settings/webhooks/remove" hx-target="#webhooks-list" hx-swap="innerHTML" hx-confirm="Are you sure you want to remove this webhook?" style="display: inline;">
-                            <input type="hidden" name="service" value="{{ webhook.service }}">
-                            <button class="button is-small is-danger" type="submit">
-                                <span class="icon">
-                                    <i class="fas fa-trash"></i>
-                                </span>
-                                <span>Remove</span>
-                            </button>
-                        </form>
-                    </div>
-                </td>
-            </tr>
-            {% endfor %}
-        </tbody>
-    </table>
-</div>
-{% else %}
-<div class="notification is-info is-light">
-    <p>No webhooks configured yet. Add your first webhook using the form below.</p>
-</div>
-{% endif %}