feat(deps): Update jacquard dependencies to 0.8.0

This commit updates the following jacquard dependencies:

* jacquard * jacquard-api * jacquard-common * jacquard-derive *
jacquard-identity * jacquard-oauth

This brings in bug fixes, new features, and performance improvements
from the latest versions of these crates.

feat(deps): Update jacquard dependencies to 0.8.0

ARCHITECTURE.md

···
 cursor = fs.readFileSync("cursor.txt") || Math.floor(Date.now() * 1000)
 
 // 2. Connect to Jetstream
-jetstream = new Jetstream({ 
+jetstream = new Jetstream({
   endpoint: "wss://jetstream1.us-east.fire.hose.cam/subscribe",
   cursor,
   wantedCollections: ["app.bsky.feed.post"]
···
 jetstream.onCreate("app.bsky.feed.post", async (event) => {
   // Extract image blobs (CIDs) from post
   const blobs = extractBlobsFromEvent(event)
-  
+
   // Enqueue for processing
   const job = { postUri, postCid, postDid, blobs, timestamp, attempts: 0 }
   await queue.enqueue(job)
···
 
 // 5. On match found, execute moderation actions
 worker.onMatchFound((postUri, postCid, postDid, match) => {
-  if (match.matchedCheck.toLabel) 
+  if (match.matchedCheck.toLabel)
     await createPostLabel(...)
   if (match.matchedCheck.reportPost)
     await createPostReport(...)
···
   // Step 1: Decode image via Sharp
   const image = sharp(buffer)
   const metadata = await image.metadata()
-  
+
   // Step 2: Resize to 8x8 grayscale
   // CRITICAL: fit: "fill" preserves aspect ratio, may add padding
   const resized = await image
···
     .grayscale()
     .raw()
     .toBuffer()
-  
+
   // Step 3: Extract pixel values (Uint8Array, 0-255 range)
   const pixels = new Uint8Array(resized)
-  
+
   // Step 4: Compute average brightness
   const avg = pixels.reduce((sum, val) => sum + val, 0) / pixels.length
-  
+
   // Step 5: Create 64-bit hash (8x8 = 64 pixels)
   let hash = ""
   for (let i = 0; i < pixels.length; i++) {
     hash += pixels[i] > avg ? "1" : "0"
   }
-  
+
   // Step 6: Convert binary string to hex (16 character string)
   return BigInt(`0b${hash}`).toString(16).padStart(16, "0")
 }
···
    - Sharp converts RGB to single channel (standard luminosity formula)
    - Range: 0-255
 
-3. **No normalization:** 
+3. **No normalization:**
    - Raw pixel values compared to mean (not normalized)
    - This is correct for perceptual hashing
 
···
   // Convert hex to BigInt
   const a = BigInt(`0x${hash1}`)
   const b = BigInt(`0x${hash2}`)
-  
+
   // XOR finds differing bits
   const xor = a ^ b
-  
+
   // Count set bits (Brian Kernighan's algorithm)
   let count = 0
   let n = xor
···
     count++
     n &= n - 1n  // Remove rightmost set bit
   }
-  
+
   return count
 }
 
 function findMatch(phash: string, checks: BlobCheck[]): BlobCheck | null {
   for (const check of checks) {
     const threshold = check.hammingThreshold ?? 5
-    
+
     for (const checkPhash of check.phashes) {
       const distance = hammingDistance(phash, checkPhash)
-      
+
       if (distance <= threshold) {
         return check  // First match wins
       }
     }
   }
-  
+
   return null
 }
 ```
···
   // GET https://plc.directory/{did}
   // Extract service with id="atproto_pds" and type="AtprotoPersonalDataServer"
   // Return serviceEndpoint URL
-  
+
   // Cached per DID to avoid repeated lookups
 }
 ```
···
   for (const checkPhash of check.phashes) {
     const distance = hammingDistance(phash, checkPhash)
     const threshold = check.hammingThreshold ?? defaultThreshold
-    
+
     if (distance <= threshold) {
       return {
         phash,
···
 ```typescript
 class MetricsCollector {
   counters: Map<string, number>
-  
+
   increment(metric: string, value?: number)
   get(metric: string): number
   getAll(): Record<string, number>
···
 8. **Graceful shutdown** saves cursor and exits cleanly
 
 This is a **real-time, event-driven system** - no polling, no batch processing. Every post on Bluesky is potentially seen and processed within seconds.
-

Cargo.lock

···
 
 [[package]]
 name = "jacquard"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "bon",
  "bytes",
···
 
 [[package]]
 name = "jacquard-api"
-version = "0.7.1"
+version = "0.8.0"
 dependencies = [
  "bon",
  "bytes",
···
 
 [[package]]
 name = "jacquard-common"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "base64 0.22.1",
  "bon",
···
 
 [[package]]
 name = "jacquard-derive"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "proc-macro2",
  "quote",
···
 
 [[package]]
 name = "jacquard-identity"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "bon",
  "bytes",
···
 
 [[package]]
 name = "jacquard-oauth"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "base64 0.22.1",
  "bytes",

Cargo.toml

···
 [package]
 name = "skywatch-phash-rs"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2024"
 authors = ["Giulia <[e-mail]"]
 description = "Perceptual hash-based image moderation for Bluesky (Rust rewrite)"

rules/blobs.json

···
       "0f093139797b7967",
       "fdedc3030100c0fd",
       "0f7f707dcc0c0600",
-      "87030303199dff81"
+      "87030303199dff81",
+      "3f17070707077f7e",
+      "f95d1f1ffcf8fbff"
     ],
     "label": "troll",
     "comment": "Image is used in harrassment campaign targeting Will Stancil",
···
     "ignoreDID": ["did:plc:7umvpuxe2vbrc3zrzuquzniu"]
   },
   {
-    "phashes": ["00fffd7cd8da5000"],
+    "phashes": ["00fffd7cd8da5000", "ffbf8f83999b9b00", "00ffe300ff8000ff"],
     "label": "maga-trump",
     "comment": "Pro-trump imagery",
     "reportAcct": true,

src/agent/session.rs

···
 
         tracing::info!("Successfully logged in as {} ({})", auth.handle, auth.did);
 
-        let did: Arc<str> = Arc::from(auth.did.to_string());
+        let did = Arc::from(auth.did.as_str());
         let agent = Arc::new(Agent::from(session));
 
         Ok(Self { agent, did })

src/cache/mod.rs

···
         F: FnOnce() -> Fut,
         Fut: std::future::Future<Output = Result<String>>,
     {
-        // Try to get from cache
         if let Some(cached) = self.get(cid).await? {
             return Ok(cached);
         }
 
-        // Compute if not cached
         let phash = compute_fn().await?;
-
-        // Store in cache
         self.set(cid, &phash).await?;
 
         Ok(phash)

src/jetstream/events.rs

···
-use jacquard_common::types::string::{AtprotoStr, Cid};
+use jacquard_common::types::string::Cid;
 use jacquard_common::types::value::Data;
 use miette::Result;
 
···
 
 /// Extract blob references from a post record
 ///
-/// Handles two cases:
-/// 1. Direct images: record.embed.images[].image.ref.$link
-/// 2. Quote posts with media: record.embed.media.images[].image.ref.$link
+/// Handles:
+/// 1. Direct images: record.embed.images[].image (Blob type)
+/// 2. Quote posts with media: record.embed.media.images[].image (Blob type)
+/// 3. External link cards: record.embed.external.thumb (Blob type)
+///
+/// TODO: Add support for app.bsky.embed.video (embed.video blob)
 pub fn extract_blobs_from_record(record: &Data) -> Result<Vec<BlobReference>> {
     let mut blobs = Vec::new();
 
···
         }
     }
 
+    // Case 3: External link card thumbnail (embed.external.thumb)
+    if let Some(Data::Object(external)) = embed.0.get("external") {
+        if let Some(blob_ref) = extract_blob_direct(external.0.get("thumb")) {
+            blobs.push(blob_ref);
+        }
+    }
+
     Ok(blobs)
 }
 
-/// Extract a single blob reference from an image object
+/// Extract a single blob reference from an image object (embed.images[].image)
 fn extract_blob_from_image(img: &Data) -> Option<BlobReference> {
     use jacquard_common::IntoStatic;
+    use jacquard_common::types::blob::Blob;
 
     let Data::Object(img_obj) = img else {
         return None;
     };
 
-    let Data::Object(image_obj) = img_obj.0.get("image")? else {
+    // The image field is directly a Blob type in jacquard
+    let Data::Blob(Blob { r#ref, mime_type, .. }) = img_obj.0.get("image")? else {
         return None;
     };
 
-    let Data::Object(ref_obj) = image_obj.0.get("ref")? else {
-        return None;
-    };
+    Some(BlobReference {
+        cid: Cid::str(r#ref.as_str()).into_static(),
+        mime_type: Some(mime_type.to_string().into()),
+    })
+}
+
+/// Extract a blob reference directly from a Data::Blob (for external.thumb, etc.)
+fn extract_blob_direct(data: Option<&Data>) -> Option<BlobReference> {
+    use jacquard_common::IntoStatic;
+    use jacquard_common::types::blob::Blob;
 
-    let Data::String(AtprotoStr::String(cid_str)) = ref_obj.0.get("$link")? else {
+    let Data::Blob(Blob { r#ref, mime_type, .. }) = data? else {
         return None;
     };
 
-    let mime_type = image_obj
-        .0
-        .get("mimeType")
-        .and_then(|v| match v {
-            Data::String(AtprotoStr::String(s)) => Some(s.to_string().into()),
-            _ => None,
-        });
-
     Some(BlobReference {
-        cid: Cid::str(cid_str.as_str()).into_static(),
-        mime_type,
+        cid: Cid::str(r#ref.as_str()).into_static(),
+        mime_type: Some(mime_type.to_string().into()),
     })
 }
 

src/jetstream/mod.rs

···
+use futures::StreamExt;
+use jacquard_common::IntoStatic;
 use jacquard_common::jetstream::{CommitOperation, JetstreamMessage, JetstreamParams};
 use jacquard_common::types::string::AtUri;
 use jacquard_common::xrpc::{SubscriptionClient, TungsteniteSubscriptionClient};
-use jacquard_common::IntoStatic;
 use miette::{IntoDiagnostic, Result};
-use futures::StreamExt;
 use tokio::sync::mpsc;
 use tracing::{debug, error, info, warn};
 use url::Url;
···
         let mut message_count = 0u64;
         let mut last_cursor: Option<i64> = None;
         let mut cursor_update_interval = tokio::time::interval(std::time::Duration::from_secs(10));
+        let mut heartbeat_interval = tokio::time::interval(std::time::Duration::from_secs(30));
 
         loop {
             tokio::select! {
-                Some(result) = messages.next() => {
-                    match result {
-                        Ok(msg) => {
-                            message_count += 1;
-                            if message_count % 1000 == 0 {
-                                debug!("Processed {} messages", message_count);
-                            }
+                _ = heartbeat_interval.tick() => {
+                    info!("Jetstream heartbeat: {} messages received so far", message_count);
+                }
+                msg_option = messages.next() => {
+                    match msg_option {
+                        Some(result) => {
+                            match result {
+                                Ok(msg) => {
+                                    message_count += 1;
+                                    if message_count % 100 == 0 {
+                                        debug!("Processed {} messages", message_count);
+                                    }
+                                    if message_count == 1 {
+                                        info!("First message received!");
+                                    }
 
-                            // Extract cursor from message
-                            let cursor = match &msg {
-                                JetstreamMessage::Commit { time_us, .. } => Some(*time_us),
-                                JetstreamMessage::Identity { time_us, .. } => Some(*time_us),
-                                JetstreamMessage::Account { time_us, .. } => Some(*time_us),
-                            };
+                                    // Extract cursor from message
+                                    let cursor = match &msg {
+                                        JetstreamMessage::Commit { time_us, .. } => Some(*time_us),
+                                        JetstreamMessage::Identity { time_us, .. } => Some(*time_us),
+                                        JetstreamMessage::Account { time_us, .. } => Some(*time_us),
+                                    };
 
-                            if let Some(c) = cursor {
-                                last_cursor = Some(c);
-                            }
+                                    if let Some(c) = cursor {
+                                        last_cursor = Some(c);
+                                    }
 
-                            if let Err(e) = self.process_message(msg, &job_sender) {
-                                error!("Error processing message: {}", e);
+                                    if let Err(e) = self.process_message(msg, &job_sender) {
+                                        error!("Error processing message: {}", e);
+                                    }
+                                }
+                                Err(e) => {
+                                    error!("Jetstream error: {}", e);
+                                }
                             }
                         }
-                        Err(e) => {
-                            error!("Jetstream error: {}", e);
+                        None => {
+                            warn!("Jetstream stream ended unexpectedly");
+                            break;
                         }
                     }
                 }
···
             } => {
                 // Only process create operations on posts
                 if commit.collection.as_ref() != "app.bsky.feed.post" {
+                    debug!("Skipping non-post collection: {}", commit.collection);
                     return Ok(());
                 }
 
                 if !matches!(commit.operation, CommitOperation::Create) {
+                    debug!("Skipping non-create operation");
                     return Ok(());
                 }
 
                 // Parse record to extract blobs (skip if no record)
                 let Some(record_data) = &commit.record else {
+                    debug!("Skipping post with no record");
                     return Ok(());
                 };
 
                 let blobs = events::extract_blobs_from_record(record_data)?;
 
                 if blobs.is_empty() {
+                    debug!("Post has no blobs");
                     return Ok(());
                 }
 
+                debug!("Found post with {} blob(s)!", blobs.len());
+
                 let post_uri_str = format!("at://{}/{}/{}", did, commit.collection, commit.rkey);
-                let post_uri = AtUri::new(&post_uri_str)
-                    .into_diagnostic()?
-                    .into_static();
+                let post_uri = AtUri::new(&post_uri_str).into_diagnostic()?.into_static();
 
-                debug!(
-                    "Post with {} blob(s): {}",
-                    blobs.len(),
-                    post_uri
-                );
+                debug!("Post with {} blob(s): {}", blobs.len(), post_uri);
 
                 // Create job
                 let post_cid = commit
···
                 };
 
                 // Send to queue
-                if let Err(e) = job_sender.send(job) {
-                    warn!("Failed to send job to queue: {}", e);
+                match job_sender.send(job) {
+                    Ok(_) => {
+                        debug!("Sent job to channel: {}", post_uri_str);
+                        Ok(())
+                    }
+                    Err(e) => {
+                        warn!("Failed to send job to queue: {}", e);
+                        Ok(())
+                    }
                 }
-
-                Ok(())
             }
             JetstreamMessage::Identity { .. } => {
                 // Ignore identity updates for now

src/main.rs

···
 use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::time::interval;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use skywatch_phash_rs::{
     agent::AgentSession,
···
         blob_checks.clone(),
         metrics.clone(),
     );
-    info!("Worker pool created with {} workers", config.processing.concurrency);
+    info!(
+        "Worker pool created with {} workers",
+        config.processing.concurrency
+    );
 
     // Create shutdown channels
     let (_shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
···
 
     // Start job receiver (receives from jetstream, pushes to queue)
     info!("Starting job receiver...");
-    let mut queue_for_receiver = queue.clone();
     let receiver_metrics = metrics.clone();
+    let receiver_config = config.clone();
     let receiver_handle = tokio::spawn(async move {
+        // Create fresh queue connection for receiver
+        let mut queue_for_receiver = match JobQueue::new(&receiver_config).await {
+            Ok(q) => q,
+            Err(e) => {
+                error!("Failed to create queue for receiver: {}", e);
+                return;
+            }
+        };
+
         while let Some(job) = job_rx.recv().await {
+            debug!("Job receiver got job: {}", job.post_uri);
             receiver_metrics.inc_jobs_received();
-            if let Err(e) = queue_for_receiver.push(&job).await {
-                error!("Failed to push job to queue: {}", e);
-                receiver_metrics.inc_jobs_failed();
+            match queue_for_receiver.push(&job).await {
+                Ok(_) => {
+                    debug!("Pushed job to Redis queue: {}", job.post_uri);
+                }
+                Err(e) => {
+                    error!("Failed to push job to queue: {}", e);
+                    receiver_metrics.inc_jobs_failed();
+                }
             }
         }
         info!("Job receiver stopped");
···
 
         let worker_future = async move {
             info!("Worker {} starting", worker_id);
-            if let Err(e) = worker_pool_clone.start(queue_clone, cache_clone, worker_shutdown).await {
+            if let Err(e) = worker_pool_clone
+                .start(queue_clone, cache_clone, worker_shutdown)
+                .await
+            {
                 error!("Worker {} failed: {}", worker_id, e);
             }
             info!("Worker {} stopped", worker_id);

src/moderation/account.rs

···
 use jacquard_api::tools_ozone::moderation::{
     ModEventLabel, ModEventReport, ModEventTakedown, ModTool,
 };
-use jacquard_common::CowStr;
-use jacquard_common::xrpc::XrpcClient;
+use jacquard_common::{CowStr, IntoStatic};
+use jacquard_common::types::string::{AtUri, Did};
 
-use miette::{IntoDiagnostic, Result};
+use miette::Result;
 use std::collections::BTreeMap;
 use tracing::{debug, info};
 
 use crate::config::Config;
 use crate::moderation::rate_limiter::RateLimiter;
 use crate::moderation::{
-    build_mod_tool_meta, build_moderation_call_opts, build_timestamped_comment, parse_did,
+    build_mod_tool_meta, build_timestamped_comment, send_moderation_event,
 };
 
 /// Label an account with a specific label via Ozone moderation API
-pub async fn label_account(
+pub async fn label_account<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    did: &str,
+    did: &Did<'a>,
     label_val: &str,
     check_comment: &str,
-    post_uri: &str,
+    post_uri: &AtUri<'a>,
     phash: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Labeling account {} with label: {}", did, label_val);
 
     let comment = build_timestamped_comment(check_comment);
-    let meta = build_mod_tool_meta(post_uri, phash);
+    let meta = build_mod_tool_meta(post_uri.as_str(), phash);
 
-    // Create moderation label event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventLabel(Box::new(
             ModEventLabel::builder()
                 .create_label_vals(vec![CowStr::from(label_val)])
···
         )))
         .subject(EmitEventSubject::RepoRef(Box::new(
             RepoRef::builder()
-                .did(parse_did(did)?)
+                .did(did.clone().into_static())
                 .build(),
         )))
         .mod_tool(ModTool {
···
         })
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully labeled account: {}", did);
 
···
 }
 
 /// Report an account to ozone moderation
-pub async fn report_account(
+pub async fn report_account<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    did: &str,
+    did: &Did<'a>,
     reason: ReasonType<'static>,
     check_comment: &str,
-    post_uri: &str,
+    post_uri: &AtUri<'a>,
     phash: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Reporting account {} to ozone: {:?}", did, reason);
 
     let comment = build_timestamped_comment(check_comment);
-    let meta = build_mod_tool_meta(post_uri, phash);
+    let meta = build_mod_tool_meta(post_uri.as_str(), phash);
 
-    // Create moderation report event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventReport(Box::new(
             ModEventReport::builder()
                 .report_type(reason)
···
         )))
         .subject(EmitEventSubject::RepoRef(Box::new(
             RepoRef::builder()
-                .did(parse_did(did)?)
+                .did(did.clone().into_static())
                 .build(),
         )))
         .subject_blob_cids(vec![])
···
         })
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully reported account: {}", did);
 
···
 }
 
 /// Takedown an account via Ozone moderation API
-pub async fn takedown_account(
+pub async fn takedown_account<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    did: &str,
+    did: &Did<'a>,
     comment: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Taking down account: {}", did);
 
-    // Create moderation takedown event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventTakedown(Box::new(
             ModEventTakedown {
                 comment: Some(CowStr::from(comment)),
···
         )))
         .subject(EmitEventSubject::RepoRef(Box::new(
             RepoRef::builder()
-                .did(parse_did(did)?)
+                .did(did.clone().into_static())
                 .build(),
         )))
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully took down account: {}", did);
 

src/moderation/helpers.rs

···
+use jacquard::client::{Agent, MemoryCredentialSession};
+use jacquard_api::tools_ozone::moderation::emit_event::EmitEvent;
 use jacquard_common::types::string::{AtprotoStr, Did};
 use jacquard_common::types::value::{Data, Object};
-use jacquard_common::xrpc::CallOptions;
+use jacquard_common::xrpc::{CallOptions, XrpcClient};
 use jacquard_common::CowStr;
 use jacquard_common::smol_str::SmolStr;
 use jacquard_common::IntoStatic;
···
 use std::collections::BTreeMap;
 
 use crate::config::Config;
+use crate::moderation::rate_limiter::RateLimiter;
 
 pub fn build_timestamped_comment(check_comment: &str) -> String {
     let timestamp = chrono::Utc::now().to_rfc3339();
···
         .into_diagnostic()
         .map(|did| did.into_static())
 }
+
+pub async fn send_moderation_event<'a>(
+    agent: &Agent<MemoryCredentialSession>,
+    config: &Config,
+    rate_limiter: &RateLimiter,
+    event: EmitEvent<'a>,
+) -> Result<()> {
+    rate_limiter.wait().await;
+
+    let opts = build_moderation_call_opts(config);
+    agent.send_with_opts(event, opts).await.into_diagnostic()?;
+
+    Ok(())
+}

src/moderation/post.rs

···
 use jacquard_api::tools_ozone::moderation::{
     ModEventLabel, ModEventReport, ModEventTakedown, ModTool,
 };
-use jacquard_common::CowStr;
-use jacquard_common::types::string::{AtUri, Cid};
-use jacquard_common::xrpc::XrpcClient;
-use miette::{IntoDiagnostic, Result};
+use jacquard_common::{CowStr, IntoStatic};
+use jacquard_common::types::string::{AtUri, Cid, Did};
+use miette::Result;
 use std::collections::BTreeMap;
 use tracing::{debug, info};
 
 use crate::config::Config;
 use crate::moderation::rate_limiter::RateLimiter;
 use crate::moderation::{
-    build_mod_tool_meta, build_moderation_call_opts, build_timestamped_comment, parse_did,
+    build_mod_tool_meta, build_timestamped_comment, send_moderation_event,
 };
 
 /// Label a post with a specific label via Ozone moderation API
-pub async fn label_post(
+pub async fn label_post<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    post_uri: &str,
-    post_cid: &str,
+    post_uri: &AtUri<'a>,
+    post_cid: &Cid<'a>,
     label_val: &str,
     check_comment: &str,
     phash: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Labeling post {} with label: {}", post_uri, label_val);
 
     let comment = build_timestamped_comment(check_comment);
-    let meta = build_mod_tool_meta(post_uri, phash);
+    let meta = build_mod_tool_meta(post_uri.as_str(), phash);
 
-    // Create moderation label event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventLabel(Box::new(
             ModEventLabel::builder()
                 .create_label_vals(vec![CowStr::from(label_val)])
···
         )))
         .subject(EmitEventSubject::StrongRef(Box::new(
             StrongRef::builder()
-                .uri(AtUri::new(post_uri).into_diagnostic()?)
-                .cid(Cid::str(post_cid))
+                .uri(post_uri.clone().into_static())
+                .cid(post_cid.clone().into_static())
                 .build(),
         )))
         .mod_tool(ModTool {
···
         })
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully labeled post: {}", post_uri);
 
···
 }
 
 /// Report a post to ozone moderation
-pub async fn report_post(
+pub async fn report_post<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    post_uri: &str,
-    _post_cid: &str,
+    post_uri: &AtUri<'a>,
+    _post_cid: &Cid<'a>,
+    post_did: &Did<'a>,
     reason: ReasonType<'static>,
     check_comment: &str,
     phash: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Reporting post {} to ozone: {:?}", post_uri, reason);
 
     let comment = build_timestamped_comment(check_comment);
-    let meta = build_mod_tool_meta(post_uri, phash);
-    let did_str = extract_did_from_uri(post_uri)?;
+    let meta = build_mod_tool_meta(post_uri.as_str(), phash);
 
-    // Create moderation report event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventReport(Box::new(
             ModEventReport::builder()
                 .report_type(reason)
···
         )))
         .subject(EmitEventSubject::RepoRef(Box::new(
             RepoRef::builder()
-                .did(parse_did(&did_str)?)
+                .did(post_did.clone().into_static())
                 .build(),
         )))
         .subject_blob_cids(vec![])
···
         })
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully reported post: {}", post_uri);
 
···
 }
 
 /// Takedown a post via Ozone moderation API
-pub async fn takedown_post(
+pub async fn takedown_post<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
-    post_uri: &str,
-    post_cid: &str,
+    post_uri: &AtUri<'a>,
+    post_cid: &Cid<'a>,
     comment: &str,
-    created_by: &str,
+    created_by: &Did<'a>,
 ) -> Result<()> {
-    // Wait for rate limiter before making request
-    rate_limiter.wait().await;
-
     info!("Taking down post: {}", post_uri);
 
-    // Create moderation takedown event using jacquard-api types
     let event = EmitEvent::new()
-        .created_by(parse_did(created_by)?)
+        .created_by(created_by.clone().into_static())
         .event(EmitEventEvent::ModEventTakedown(Box::new(
             ModEventTakedown {
                 comment: Some(CowStr::from(comment)),
···
         )))
         .subject(EmitEventSubject::StrongRef(Box::new(
             StrongRef::builder()
-                .uri(AtUri::new(post_uri).into_diagnostic()?)
-                .cid(Cid::str(post_cid))
+                .uri(post_uri.clone().into_static())
+                .cid(post_cid.clone().into_static())
                 .build(),
         )))
         .build();
 
-    let opts = build_moderation_call_opts(config);
-
-    // Send request via jacquard agent
-    let _response = agent.send_with_opts(event, opts).await.into_diagnostic()?;
+    send_moderation_event(agent, config, rate_limiter, event).await?;
 
     debug!("Successfully took down post: {}", post_uri);
 
     Ok(())
 }
 
-/// Parse an AT URI into its components
-/// Format: at://did:plc:xxx/app.bsky.feed.post/rkey
-fn parse_at_uri(uri: &str) -> Result<(String, String, String)> {
-    let uri = uri
-        .strip_prefix("at://")
-        .ok_or_else(|| miette::miette!("Invalid AT URI format: missing 'at://' prefix"))?;
-
-    let parts: Vec<&str> = uri.split('/').collect();
-    if parts.len() != 3 {
-        return Err(miette::miette!(
-            "Invalid AT URI format: expected 3 parts, got {}",
-            parts.len()
-        ));
-    }
-
-    Ok((
-        parts[0].to_string(), // repo (DID)
-        parts[1].to_string(), // collection
-        parts[2].to_string(), // rkey
-    ))
-}
-
-/// Extract DID from AT URI
-fn extract_did_from_uri(uri: &str) -> Result<String> {
-    let (did, _, _) = parse_at_uri(uri)?;
-    Ok(did)
-}
-
 #[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_parse_at_uri() {
-        let uri = "at://did:plc:xyz123/app.bsky.feed.post/abc456";
-        let (repo, collection, rkey) = parse_at_uri(uri).unwrap();
-        assert_eq!(repo, "did:plc:xyz123");
-        assert_eq!(collection, "app.bsky.feed.post");
-        assert_eq!(rkey, "abc456");
-    }
-
-    #[test]
-    fn test_parse_at_uri_invalid() {
-        let uri = "https://example.com/post/123";
-        assert!(parse_at_uri(uri).is_err());
-    }
-
-    #[test]
-    fn test_extract_did_from_uri() {
-        let uri = "at://did:plc:xyz123/app.bsky.feed.post/abc456";
-        let did = extract_did_from_uri(uri).unwrap();
-        assert_eq!(did, "did:plc:xyz123");
-    }
-}
+mod tests {}

src/processor/matcher.rs

···
     default_threshold: u32,
 ) -> Option<MatchResult> {
     for check in blob_checks {
-        // Check if DID is in ignore list
         if let Some(ignore_list) = &check.ignore_did {
             if ignore_list.iter().any(|ignored_did| ignored_did.as_str() == did) {
                 debug!("Skipping check '{}' for ignored DID: {}", check.label, did);
···
 
         let threshold = check.hamming_threshold.unwrap_or(default_threshold);
 
-        // Check each phash in the check
         for check_phash in &check.phashes {
             match phash::hamming_distance(phash, check_phash.as_str()) {
                 Ok(distance) => {
···
     did: &str,
     blob: &BlobReference,
 ) -> Result<Option<MatchResult>> {
-    // Download the blob
     let image_bytes = download_blob(client, config, did, &blob.cid).await?;
-
-    // Compute phash
     let phash = phash::compute_phash(&image_bytes)?;
     debug!("Computed phash for blob {}: {}", blob.cid, phash);
 
-    // Match against checks
     let match_result = match_phash(&phash, blob_checks, did, config.phash.default_hamming_threshold);
 
     Ok(match_result)

src/processor/phash.rs

···
 use image::DynamicImage;
 use image_hasher::{HashAlg, HasherConfig, ImageHash};
-use miette::{Diagnostic, IntoDiagnostic, Result};
+use miette::Diagnostic;
 use thiserror::Error;
 
 #[derive(Debug, Error, Diagnostic)]
 pub enum PhashError {
-    #[error("Failed to decode image: {0}")]
-    ImageDecodeError(String),
-
-    #[error("Failed to compute hash: {0}")]
-    HashComputeError(String),
+    #[error("Failed to decode image")]
+    ImageDecode(#[from] image::ImageError),
 
     #[error("Invalid hash format: {0}")]
     InvalidHashFormat(String),
+
+    #[error("Invalid hex string")]
+    ParseInt(#[from] std::num::ParseIntError),
 }
 
 /// Compute perceptual hash for an image using average hash (aHash) algorithm
···
 /// 3. Compute average pixel value
 /// 4. Create 64-bit binary: 1 if pixel > avg, 0 otherwise
 /// 5. Convert to hex string (16 chars)
-pub fn compute_phash(image_bytes: &[u8]) -> Result<String> {
-    // Decode image from bytes
-    let img = image::load_from_memory(image_bytes)
-        .into_diagnostic()
-        .map_err(|e| PhashError::ImageDecodeError(e.to_string()))?;
-
+pub fn compute_phash(image_bytes: &[u8]) -> Result<String, PhashError> {
+    let img = image::load_from_memory(image_bytes)?;
     compute_phash_from_image(&img)
 }
 
 /// Compute perceptual hash from a DynamicImage
-pub fn compute_phash_from_image(img: &DynamicImage) -> Result<String> {
+pub fn compute_phash_from_image(img: &DynamicImage) -> Result<String, PhashError> {
     // Configure hasher with aHash (Mean) algorithm and 8x8 size
     let hasher = HasherConfig::new()
         .hash_alg(HashAlg::Mean) // average hash
···
 }
 
 /// Convert ImageHash to hex string format (16 chars, matching TS output)
-fn hash_to_hex(hash: &ImageHash) -> Result<String> {
+fn hash_to_hex(hash: &ImageHash) -> Result<String, PhashError> {
     // Get hash bytes
     let bytes = hash.as_bytes();
 
···
 /// Compute hamming distance between two phash hex strings
 ///
 /// Uses Brian Kernighan's algorithm to count set bits
-pub fn hamming_distance(hash1: &str, hash2: &str) -> Result<u32> {
+pub fn hamming_distance(hash1: &str, hash2: &str) -> Result<u32, PhashError> {
     // Validate input lengths
     if hash1.len() != 16 || hash2.len() != 16 {
         return Err(PhashError::InvalidHashFormat(format!(
···
         .into());
     }
 
-    // Parse hex strings to u64
-    let a = u64::from_str_radix(hash1, 16)
-        .into_diagnostic()
-        .map_err(|_| PhashError::InvalidHashFormat(format!("Invalid hex string: {}", hash1)))?;
-
-    let b = u64::from_str_radix(hash2, 16)
-        .into_diagnostic()
-        .map_err(|_| PhashError::InvalidHashFormat(format!("Invalid hex string: {}", hash2)))?;
+    let a = u64::from_str_radix(hash1, 16)?;
+    let b = u64::from_str_radix(hash2, 16)?;
 
     // XOR to find differing bits
     let xor = a ^ b;

src/queue/worker.rs

···
-use miette::Result;
+use jacquard::client::{Agent, MemoryCredentialSession};
+use jacquard_api::com_atproto::moderation::ReasonType;
+use jacquard_common::types::string::Did;
+use miette::{IntoDiagnostic, Result};
 use reqwest::Client;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::sleep;
-use tracing::{error, info};
-use jacquard::client::{Agent, MemoryCredentialSession};
-use jacquard_api::com_atproto::moderation::ReasonType;
+use tracing::{debug, error, info};
 
 use crate::agent::AgentSession;
 use crate::cache::PhashCache;
···
                 info!(concat!($action_name, " completed for: {}"), $subject);
             } else {
                 $metrics_skip;
-                info!(concat!($action_name, " already done, skipping: {}"), $subject);
+                info!(
+                    concat!($action_name, " already done, skipping: {}"),
+                    $subject
+                );
             }
         }
     };
···
                 info!(concat!($action_name, " completed for: {}"), $subject);
             } else {
                 $metrics_skip;
-                info!(concat!($action_name, " already done, skipping: {}"), $subject);
+                info!(
+                    concat!($action_name, " already done, skipping: {}"),
+                    $subject
+                );
             }
         }
     };
···
                 job_result = queue.pop(1) => {
                     match job_result {
                         Ok(Some(job)) => {
-                            // Create new redis connection for this job
+                            debug!("Worker popped job from queue: {}", job.post_uri);
                             let redis_client = match redis::Client::open(self.config.redis.url.as_str()) {
                                 Ok(c) => c,
                                 Err(e) => {
···
                                 }
                             };
 
-                            // Process job
+                            let job_clone = job.clone();
                             if let Err(e) = Self::process_job(
                                 &self.config,
                                 &self.client,
···
                                 &mut cache,
                                 &mut redis_conn,
                                 job,
-                                &mut queue,
                                 self.agent.did(),
                             )
                             .await
                             {
                                 error!("Worker task failed: {}", e);
+                                self.metrics.inc_jobs_failed();
+                                if let Err(retry_err) = queue.retry(job_clone).await {
+                                    error!("Failed to retry job: {}", retry_err);
+                                }
                             }
                         }
                         Ok(None) => {
···
         cache: &mut PhashCache,
         redis_conn: &mut redis::aio::MultiplexedConnection,
         job: ImageJob,
-        queue: &mut JobQueue,
         created_by: &str,
     ) -> Result<()> {
-        info!("Processing job: {}", job.post_uri);
+        debug!("Processing job: {}", job.post_uri);
 
-        // Process all blobs and find matches
-        let matches = Self::process_job_blobs(config, client, blob_checks, metrics, cache, &job).await?;
+        let matches =
+            Self::process_job_blobs(config, client, blob_checks, metrics, cache, &job).await?;
 
         if matches.is_empty() {
-            info!("No matches found for job: {}", job.post_uri);
+            debug!("No matches found for job: {}", job.post_uri);
             metrics.inc_jobs_processed();
             return Ok(());
         }
 
         // Take moderation actions for each match
         for match_result in matches {
-            if let Err(e) =
-                Self::take_moderation_action(config, agent, metrics, rate_limiter, redis_conn, &job, &match_result, created_by)
-                    .await
+            if let Err(e) = Self::take_moderation_action(
+                config,
+                agent,
+                metrics,
+                rate_limiter,
+                redis_conn,
+                &job,
+                &match_result,
+                created_by,
+            )
+            .await
             {
                 error!("Failed to take moderation action: {}", e);
-                metrics.inc_jobs_failed();
-                // Retry the job
-                queue.retry(job.clone()).await?;
+                // Don't retry here - worker will handle it
                 return Err(e);
             }
         }
 
-        info!("Successfully processed job: {}", job.post_uri);
+        debug!("Successfully processed job: {}", job.post_uri);
         metrics.inc_jobs_processed();
 
         Ok(())
···
                 metrics.inc_blobs_downloaded();
 
                 // Download and compute
-                let image_bytes = matcher::download_blob(client, config, &job.post_did, &blob.cid).await?;
+                let image_bytes =
+                    matcher::download_blob(client, config, &job.post_did, &blob.cid).await?;
                 let computed_phash = crate::processor::phash::compute_phash(&image_bytes)?;
 
                 // Store in cache
···
         created_by: &str,
     ) -> Result<()> {
         let check = &match_result.matched_check;
+        let created_by_did = Did::new(created_by).into_diagnostic()?;
 
         info!(
             "Taking moderation action for label '{}' on post: {}",
···
                 rate_limiter,
                 &job.post_uri,
                 &job.post_cid,
+                &job.post_did,
                 ReasonType::ComAtprotoModerationDefsReasonSpam,
                 &check.comment,
                 &match_result.phash,
-                created_by,
+                &created_by_did,
             ),
             metrics.inc_posts_reported(),
             metrics.inc_posts_already_reported(),
···
                 &check.label,
                 &check.comment,
                 &match_result.phash,
-                created_by,
+                &created_by_did,
             ),
             claims::set_label(redis_conn, &job.post_uri, &check.label, None),
             metrics.inc_posts_labeled(),
···
                 &check.comment,
                 &job.post_uri,
                 &match_result.phash,
-                created_by,
+                &created_by_did,
             ),
             metrics.inc_accounts_reported(),
             metrics.inc_accounts_already_reported(),
···
                 &check.comment,
                 &job.post_uri,
                 &match_result.phash,
-                created_by,
+                &created_by_did,
             ),
             claims::set_label(redis_conn, &job.post_did, &check.label, None),
             metrics.inc_accounts_labeled(),

src/types/mod.rs

···
 where
     D: Deserializer<'de>,
 {
-    let s = String::deserialize(deserializer)?;
-    Ok(CowStr::from(s))
+    let s: &str = Deserialize::deserialize(deserializer)?;
+    Ok(CowStr::from(s).into_static())
 }
 
 fn deserialize_cowstr_vec<'de, D>(deserializer: D) -> Result<Vec<CowStr<'static>>, D::Error>
 where
     D: Deserializer<'de>,
 {
-    let strings: Vec<String> = Vec::deserialize(deserializer)?;
-    Ok(strings.into_iter().map(CowStr::from).collect())
+    let strings: Vec<&str> = Vec::deserialize(deserializer)?;
+    Ok(strings.into_iter().map(|s| CowStr::from(s).into_static()).collect())
 }
 
 fn deserialize_option_cowstr<'de, D>(deserializer: D) -> Result<Option<CowStr<'static>>, D::Error>
 where
     D: Deserializer<'de>,
 {
-    let opt: Option<String> = Option::deserialize(deserializer)?;
-    Ok(opt.map(CowStr::from))
+    let opt: Option<&str> = Option::deserialize(deserializer)?;
+    Ok(opt.map(|s| CowStr::from(s).into_static()))
 }
 
 fn deserialize_cid<'de, D>(deserializer: D) -> Result<Cid<'static>, D::Error>
 where
     D: Deserializer<'de>,
 {
-    let s = String::deserialize(deserializer)?;
-    Ok(Cid::str(&s).into_static())
+    let s: &str = Deserialize::deserialize(deserializer)?;
+    Ok(Cid::str(s).into_static())
 }
 
 fn deserialize_did<'de, D>(deserializer: D) -> Result<Did<'static>, D::Error>
 where
     D: Deserializer<'de>,
 {
-    let s = String::deserialize(deserializer)?;
-    Did::new(&s)
+    let s: &str = Deserialize::deserialize(deserializer)?;
+    Did::new(s)
         .map(|d| d.into_static())
         .map_err(serde::de::Error::custom)
 }
···
 where
     D: Deserializer<'de>,
 {
-    let opt: Option<Vec<String>> = Option::deserialize(deserializer)?;
+    let opt: Option<Vec<&str>> = Option::deserialize(deserializer)?;
     match opt {
         Some(strings) => {
             let dids: Result<Vec<Did<'static>>, _> = strings
-                .iter()
+                .into_iter()
                 .map(|s| Did::new(s).map(|d| d.into_static()))
                 .collect();
             Ok(Some(dids.map_err(serde::de::Error::custom)?))
···
 where
     D: Deserializer<'de>,
 {
-    let s = String::deserialize(deserializer)?;
-    AtUri::new(&s)
+    let s: &str = Deserialize::deserialize(deserializer)?;
+    AtUri::new(s)
         .map(|u| u.into_static())
         .map_err(serde::de::Error::custom)
 }