Use AT protocol triggers

.env.example

@@ -6,5 +6,3 @@ WAVS_OPERATOR_MNEMONIC_2="they swarm fatigue suit run target refuse inside empow
 WAVS_OPERATOR_MNEMONIC_3="teach card phrase timber neither buyer fit hazard hammer melody fly neck"
 WAVS_AGGREGATOR_CREDENTIAL="test test test test test test test test test test test junk"
 DEPLOYER_PRIVATE_KEY=0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80
-
-WAVS_ENV_OPERATOR_TARGET_HANDLE="wavs.bsky.social"

backend/docker/wavs-operator.yml

@@ -5,9 +5,16 @@ services:
     stop_signal: SIGKILL
     environment:
       - WAVS_SUBMISSION_MNEMONIC=${WAVS_OPERATOR_MNEMONIC:-}
-      - WAVS_ENV_OPERATOR_TARGET_HANDLE=${WAVS_ENV_OPERATOR_TARGET_HANDLE:-}
-      - WAVS_ENV_EAS_SCHEMA_UID=${WAVS_ENV_EAS_SCHEMA_UID:-}
-    command: ["wavs", "--home", "/wavs-home", "--data", "/var/${COMPOSE_PROJECT_NAME:-}", "--port", "${COMPOSE_WAVS_PORT:-}"]
+    command:
+      [
+        "wavs",
+        "--home",
+        "/wavs-home",
+        "--data",
+        "/var/${COMPOSE_PROJECT_NAME:-}",
+        "--port",
+        "${COMPOSE_WAVS_PORT:-}",
+      ]
     volumes:
       - "${COMPOSE_WAVS_HOME:-}:/wavs-home"
     network_mode: "host"

components/operator/src/at_protocol_client.rs

@@ -3,9 +3,11 @@ use atrium_api::{
         defs::{FeedViewPost, PostViewData, ThreadViewPostRepliesItem},
         get_author_feed,
         get_post_thread::{self, OutputThreadRefs},
+        get_posts, post,
     },
     com::atproto::identity::resolve_handle,
-    types::{Object, Union},
+    com::atproto::repo::describe_repo,
+    types::{Object, TryFromUnknown, Union},
 };
 use wavs_wasi_utils::http::{fetch_json, http_request_get};
 
@@ -106,4 +108,37 @@ impl AtProtocolClient {
             _ => Err(anyhow::anyhow!("Could not parse post thread: {}", uri)),
         }
     }
+
+    /// Fetch a specific post record by URI
+    pub async fn get_post_record(&self, uri: &str) -> anyhow::Result<post::Record> {
+        let url = format!("{}/{}?uris={uri}", self.api_base, get_posts::NSID);
+
+        let request = http_request_get(&url)?;
+        let response: get_posts::Output = fetch_json(request).await?;
+
+        if let Some(post_data) = response.data.posts.into_iter().next() {
+            Ok(TryFromUnknown::try_from_unknown(post_data.data.record)?)
+        } else {
+            Err(anyhow::anyhow!("Post not found: {}", uri))
+        }
+    }
+
+    /// Check if a post is a reply to another post
+    /// Returns the parent URI if it exists, None otherwise
+    pub async fn get_parent_post_uri(&self, uri: &str) -> anyhow::Result<Option<String>> {
+        let post_record = self.get_post_record(uri).await?;
+        Ok(post_record.reply.clone().map(|r| r.parent.uri.clone()))
+    }
+
+    /// Get the handle (username) for a given DID
+    /// This converts a DID like "did:plc:..." to a username like "alice.bsky.social"
+    /// Uses the com.atproto.repo.describeRepo endpoint to get repository information
+    pub async fn fetch_handle(&self, did: &str) -> anyhow::Result<String> {
+        let describe_url = format!("{}/{}?repo={}", self.api_base, describe_repo::NSID, did);
+
+        let request = http_request_get(&describe_url)?;
+        let response: describe_repo::Output = fetch_json(request).await?;
+
+        Ok(response.data.handle.to_string())
+    }
 }

components/operator/src/lib.rs

@@ -11,7 +11,10 @@ use crate::{
     attestation::{build_attestation_payload, ReplyData},
     host::LogLevel,
     wasi::keyvalue::store::{self, Bucket},
-    wavs::{operator::input::TriggerData, types::events::TriggerDataCron},
+    wavs::{
+        operator::input::TriggerData,
+        types::events::{TriggerDataAtprotoEvent, TriggerDataCron},
+    },
 };
 
 wit_bindgen::generate!({
@@ -28,7 +31,7 @@ const POSTS_KEY: &str = "posts";
 const DID_KEY: &str = "did";
 const REPLIES_KEY_PREFIX: &str = "replies";
 const POST_TTL_NANOS: u64 = 1_209_600_000_000_000; // 2 weeks in nanos
-const POST_LIMT: u32 = 20;
+const POST_LIMIT: u32 = 20;
 
 #[derive(Serialize, Deserialize)]
 struct PostConfig {
@@ -45,9 +48,8 @@ struct Component;
 
 impl Guest for Component {
     fn run(action: TriggerAction) -> std::result::Result<Vec<WasmResponse>, String> {
-        let target_handle = std::env::var("WAVS_ENV_OPERATOR_TARGET_HANDLE").map_err(|_| {
-            "No target_handle provided (set WAVS_ENV_OPERATOR_TARGET_HANDLE)".to_string()
-        })?;
+        let target_handle =
+            host::config_var("TARGET_HANDLE").ok_or("No target_handle configured".to_string())?;
 
         match action.data {
             TriggerData::Cron(TriggerDataCron { trigger_time }) => {
@@ -65,6 +67,36 @@ impl Guest for Component {
                     .map_err(|e| e.to_string())
                 })
             }
+            TriggerData::AtprotoEvent(TriggerDataAtprotoEvent {
+                sequence: _,
+                timestamp,
+                repo,
+                collection,
+                rkey,
+                action,
+                cid: _,
+                record_data,
+            }) => {
+                let client = AtProtocolClient::new();
+                let timestamp: u64 = timestamp
+                    .try_into()
+                    .map_err(|_| "Invalid timestamp for atproto event".to_string())?;
+
+                block_on(async move {
+                    handle_atproto_event(
+                        client,
+                        target_handle,
+                        timestamp,
+                        repo,
+                        collection,
+                        rkey,
+                        action,
+                        record_data,
+                    )
+                    .await
+                    .map_err(|e| e.to_string())
+                })
+            }
             _ => unimplemented!("Trigger data variant not implemented"),
         }
     }
@@ -90,7 +122,7 @@ pub async fn generate_payload(
         &format!("Resolved target handle {} to DID {did}", &target_handle),
     );
 
-    let posts = client.fetch_posts(&did, POST_LIMT).await?;
+    let posts = client.fetch_posts(&did, POST_LIMIT).await?;
 
     let mut responses = vec![];
     let schema_uid = get_schema_uid()?;
@@ -179,9 +211,157 @@ pub async fn generate_payload(
     Ok(responses)
 }
 
+#[allow(clippy::too_many_arguments)]
+pub async fn handle_atproto_event(
+    client: AtProtocolClient,
+    target_handle: String,
+    timestamp: u64,
+    repo: String,
+    collection: String,
+    rkey: String,
+    action: String,
+    record_data: Option<String>,
+) -> anyhow::Result<Vec<WasmResponse>> {
+    let did_bucket = store::open(DID_KEY)?;
+
+    // Resolve and cache the target handle's DID for quick comparisons
+    let target_did = if let Some(cached_did) = did_bucket.get(&target_handle)? {
+        String::from_utf8(cached_did)?
+    } else {
+        let did = client.fetch_did(&target_handle).await?;
+        did_bucket.set(&target_handle, did.as_bytes())?;
+        did
+    };
+
+    // Only process post creation events
+    if collection != "app.bsky.feed.post" || action != "create" {
+        host::log(
+            LogLevel::Info,
+            &format!(
+                "Ignoring event: collection={}, action={}",
+                collection, action
+            ),
+        );
+        return Ok(vec![]);
+    }
+
+    // Construct the post URI from the event data (repo already includes full DID prefix)
+    let post_uri = format!("at://{}/app.bsky.feed.post/{}", repo, rkey);
+
+    // Parse the included record; if absent or invalid, skip without fetching
+    let current_post_record = match record_data.as_ref() {
+        Some(raw) => match serde_json::from_str::<post::Record>(raw) {
+            Ok(record) => record,
+            Err(e) => {
+                host::log(
+                    LogLevel::Error,
+                    &format!("Failed to parse record_data: {}", e),
+                );
+                return Ok(vec![]);
+            }
+        },
+        None => {
+            host::log(
+                LogLevel::Error,
+                "No record_data supplied on atproto event; skipping",
+            );
+            return Ok(vec![]);
+        }
+    };
+
+    // Check if this post has a parent (is a reply)
+    let parent_uri = match current_post_record.reply.as_ref() {
+        Some(reply) => {
+            host::log(
+                LogLevel::Debug,
+                &format!("Post is a reply to parent: {}", reply.parent.uri),
+            );
+            reply.parent.uri.clone()
+        }
+        None => {
+            host::log(LogLevel::Debug, "Post is not a reply, skipping");
+            return Ok(vec![]);
+        }
+    };
+
+    // Extract the parent author's DID from the parent post URI
+    // The URI format is: at://did:plc:.../app.bsky.feed.post/...
+    let parent_did = match extract_did_from_uri(&parent_uri) {
+        Ok(did) => did,
+        Err(e) => {
+            host::log(
+                LogLevel::Error,
+                &format!("Error extracting DID from parent URI: {}", e),
+            );
+            return Ok(vec![]);
+        }
+    };
+
+    // Check if the parent post is by the targeted DID (faster than handle lookups)
+    if parent_did != target_did {
+        return Ok(vec![]);
+    }
+
+    host::log(
+        LogLevel::Info,
+        "Parent post is by targeted handle! Creating response for reply",
+    );
+
+    // Extract the author's DID from the current post URI
+    let author_did = match extract_did_from_uri(&post_uri) {
+        Ok(did) => did,
+        Err(e) => {
+            host::log(
+                LogLevel::Error,
+                &format!("Error extracting DID from post URI: {}", e),
+            );
+            return Ok(vec![]);
+        }
+    };
+
+    // Create the reply data
+    let reply_data = ReplyData {
+        author_did,
+        reply_text: current_post_record.text.clone(),
+        reply_uri: post_uri.clone(),
+        parent_post_uri: parent_uri.clone(),
+        timestamp,
+    };
+
+    // Build and return the attestation
+    let schema_uid = get_schema_uid()?;
+    match build_attestation_payload(reply_data, schema_uid) {
+        Ok(response) => {
+            host::log(
+                LogLevel::Info,
+                &format!("Created attestation for reply to {}", target_handle),
+            );
+            Ok(vec![response])
+        }
+        Err(e) => {
+            host::log(
+                LogLevel::Error,
+                &format!("Failed to build attestation: {}", e),
+            );
+            Ok(vec![])
+        }
+    }
+}
+
+/// Extract DID from an ATProto URI
+/// URI format: at://did:.../collection/rkey
+fn extract_did_from_uri(uri: &str) -> anyhow::Result<String> {
+    let parts: Vec<&str> = uri.split('/').collect();
+    if parts.len() >= 3 && parts[0] == "at:" {
+        Ok(parts[2].to_string())
+    } else {
+        Err(anyhow::anyhow!("Invalid ATProto URI format: {}", uri))
+    }
+}
+
 fn get_schema_uid() -> anyhow::Result<FixedBytes<32>> {
-    let schema_str = std::env::var("WAVS_ENV_EAS_SCHEMA_UID")
-        .map_err(|_| anyhow::anyhow!("WAVS_ENV_EAS_SCHEMA_UID not set"))?;
+    let schema_str =
+        host::config_var("EAS_SCHEMA_UID").ok_or(anyhow::anyhow!("EAS_SCHEMA_UID not set"))?;
 
     schema_str
         .parse()

contracts/script/DeployEAS.s.sol

@@ -6,9 +6,12 @@ import {stdJson} from "forge-std/StdJson.sol";
 import {Strings} from "@openzeppelin/contracts/utils/Strings.sol";
 
 import {EAS} from "@ethereum-attestation-service/eas-contracts/contracts/EAS.sol";
-import {SchemaRegistry, ISchemaRegistry} from
-    "@ethereum-attestation-service/eas-contracts/contracts/SchemaRegistry.sol";
-import {ISchemaResolver} from "@ethereum-attestation-service/eas-contracts/contracts/resolver/ISchemaResolver.sol";
+import {
+    SchemaRegistry,
+    ISchemaRegistry
+} from "@ethereum-attestation-service/eas-contracts/contracts/SchemaRegistry.sol";
+import {ISchemaResolver} from
+    "@ethereum-attestation-service/eas-contracts/contracts/resolver/ISchemaResolver.sol";
 import {IWavsServiceManager} from "@wavs/src/eigenlayer/ecdsa/interfaces/IWavsServiceManager.sol";
 
 import {EASAttester} from "../src/eas/EASAttester.sol";
@@ -21,7 +24,9 @@ contract DeployEAS is Script {
     using stdJson for string;
     using Strings for address;
 
-    function run(address serviceManager) public {
+    function run(
+        address serviceManager
+    ) public {
         require(serviceManager != address(0), "Invalid service manager address");
 
         vm.startBroadcast();
@@ -40,31 +45,27 @@ contract DeployEAS is Script {
         console.log("EAS deployed at:", address(eas));
 
         AtSchemaResolver atSchemaResolver = new AtSchemaResolver(eas);
-        _contractsJson.serialize("AtSchemaResolver", address(atSchemaResolver).toChecksumHexString());
+        _contractsJson.serialize(
+            "AtSchemaResolver", address(atSchemaResolver).toChecksumHexString()
+        );
         console.log("AtSchemaResolver deployed at:", address(atSchemaResolver));
 
         bytes32 schemaUid = ISchemaRegistry(address(schemaRegistry)).register(
-            AtProtocolSchema.SCHEMA,
-            ISchemaResolver(address(atSchemaResolver)),
-            true
+            AtProtocolSchema.SCHEMA, ISchemaResolver(address(atSchemaResolver)), true
         );
         console.log("Schema registered with UID:", vm.toString(schemaUid));
 
-        EASAttester easAttester = new EASAttester(
-            IWavsServiceManager(serviceManager),
-            eas
-        );
-        string memory finalContractsJson = _contractsJson.serialize(
-            "EASAttester",
-            address(easAttester).toChecksumHexString()
-        );
+        EASAttester easAttester = new EASAttester(IWavsServiceManager(serviceManager), eas);
+        string memory finalContractsJson =
+            _contractsJson.serialize("EASAttester", address(easAttester).toChecksumHexString());
         console.log("EASAttester deployed at:", address(easAttester));
 
         vm.stopBroadcast();
 
         _schemaJson.serialize("uid", vm.toString(schemaUid));
         _schemaJson.serialize("definition", AtProtocolSchema.SCHEMA);
-        string memory finalSchemaJson = _schemaJson.serialize("resolver", address(atSchemaResolver).toChecksumHexString());
+        string memory finalSchemaJson =
+            _schemaJson.serialize("resolver", address(atSchemaResolver).toChecksumHexString());
 
         string memory rootJson = "root";
         rootJson.serialize("addresses", finalContractsJson);

taskfile/backend.yml

@@ -385,10 +385,6 @@ tasks:
       WAVS_SIGNING_PRIVATE_KEY:
         sh: cast wallet private-key --mnemonic "$(task backend:get-wavs-operator-mnemonic-{{.WAVS_INSTANCE}})" --mnemonic-index 1
       WAVS_LOG_LEVEL: '{{.RUST_LOG | default "info"}}'
-      WAVS_ENV_OPERATOR_TARGET_HANDLE:
-        sh: grep '^WAVS_ENV_OPERATOR_TARGET_HANDLE=' {{.REPO_ROOT}}/.env | cut -d'=' -f2 | tr -d '"'
-      WAVS_ENV_EAS_SCHEMA_UID:
-        sh: grep '^WAVS_ENV_EAS_SCHEMA_UID=' {{.REPO_ROOT}}/.env | cut -d'=' -f2 | tr -d '"'
     preconditions:
       - sh: test -n "${WAVS_OPERATOR_MNEMONIC}"
         msg: "WAVS_OPERATOR_MNEMONIC environment variable is not set for instance {{.WAVS_INSTANCE}}"