fix: rustfmt lint and narrow testnet config to stables

- Apply cargo +nightly fmt to address_reputation files + smoke test.
- Disable LayerZero WETH.e and SendOftDemo in the testnet config
  (non-stables, out of P0 scope); both stay easy to re-enable.
- Add processor-level README documenting tables, mechanics, P0
  limitations, and the testnet run command.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: larryl3u

processor/example-configs/address_reputation.testnet.yaml

@@ -36,14 +36,16 @@ server_config:
       # One row per OFT module. The module address is the FA creator (and
       # publisher of the `oft_core` Move module).
 
+      # Non-stable, disabled to keep the stables-focused graph clean. Flip
+      # `enabled: true` to include WETH.e in trace results.
       - name: layerzero_weth_e
         module_address: "0x2fa1f2914aa17d239410cb81ab46dd8fa9230272c58bc84e9e8b971eded79ca5"
         event_type: "0x2fa1f2914aa17d239410cb81ab46dd8fa9230272c58bc84e9e8b971eded79ca5::oft_core::OftReceived"
         recipient_field_path: to_address
         amount_field_path: amount_received_ld
         chain_id_field_path: src_eid
         evm_source_field_path: null   # OftReceived only carries `guid`, not the EVM sender
-        enabled: true
+        enabled: false
 
       - name: layerzero_usdt_e
         module_address: "0x9cda672762a6f88e4b608428dd063e03aaf6712f0a427923dd0f1416afa1c075"
@@ -75,16 +77,15 @@ server_config:
         evm_source_field_path: null
         enabled: false
 
-      # LayerZero SendOftDemo (SOD). Demo / test OFT, kept for completeness;
-      # safe to leave on since it's a real OFT-style bridge.
+      # SendOftDemo (SOD) is a LayerZero test/demo OFT, not a stable. Disabled.
       - name: layerzero_send_oft_demo
         module_address: "0xe024846967ef05baed04aa2396802b360dcd975fc704bfd261089cf5355669ad"
         event_type: "0xe024846967ef05baed04aa2396802b360dcd975fc704bfd261089cf5355669ad::oft_core::OftReceived"
         recipient_field_path: to_address
         amount_field_path: amount_received_ld
         chain_id_field_path: src_eid
         evm_source_field_path: null
-        enabled: true
+        enabled: false
 
   transaction_stream_config:
     indexer_grpc_data_service_address: "https://grpc.testnet.movementnetwork.xyz:443"

processor/src/processors/address_reputation/README.md

@@ -0,0 +1,79 @@
+# address_reputation processor
+
+Builds an owner-keyed transfer graph on Movement and annotates bridge deposits
+as terminal "head" nodes, so any address's funds can be backtraced to the bridge
+they came from.
+
+**Scope: stables + FA v2 only.** Tracks USDCx (Circle) and USDC.e / USDT.e
+(LayerZero). All in-scope tokens are FA v2 — coin v1 and non-stable assets are
+intentionally out of scope for P0.
+
+## Tables
+
+- `address_transfer_edges` — one row per FA transfer; `from`/`to` are owners
+- `bridge_inflows` — head nodes (bridge module → recipient, with `src_chain_id`)
+- `evm_address_risk_scores` — external input, NULL `evm_source` allowed
+- `address_reputation` — materialized score, `max(current, decay * sender)` upsert
+
+See `db/queries/trace.sql` for the recursive CTE that walks the graph backward,
+short-circuiting on `is_bridge_inflow`.
+
+## Bridge config (YAML, not SQL)
+
+Bridges live in `AddressReputationConfig.bridges` — mainnet/testnet differ by
+config file, not migration. Restart picks up changes.
+
+Testnet seed (`example-configs/address_reputation.testnet.yaml`), each verified
+against captured on-chain events:
+
+| name | module | event |
+|---|---|---|
+| `circle_usdcx` | `0x989577...` | `::usdcx::Mint` |
+| `layerzero_weth_e` | `0x2fa1f2...` | `::oft_core::OftReceived` |
+| `layerzero_usdt_e` | `0x9cda67...` | `::oft_core::OftReceived` |
+| `layerzero_usdc_e` | `0x339873...` | `::oft_core::OftReceived` |
+| `layerzero_usdc_e_alt` | `0xdbfc7b...` | dormant, disabled |
+| `layerzero_send_oft_demo` | `0xe02484...` | `::oft_core::OftReceived` |
+
+## Correctness mechanics
+
+- **Owner resolution**: per-txn scan of `WriteResource` for `0x1::object::ObjectCore`
+  builds `storage_id → owner`. Mirrors `fungible_asset_processor_helpers::parse_v2_coin`
+  Loop 1. Without this, bridge mints (owner-keyed) wouldn't connect to subsequent
+  user transfers (storage_id-keyed).
+- **Asset resolution**: per-txn scan of `0x1::fungible_asset::FungibleStore` builds
+  `storage_id → metadata` so every edge carries `asset_type`.
+- **Synthetic bridge edge**: bridge mints emit only a `Deposit` (no paired
+  `Withdraw`); extractor synthesizes a `bridge_module → recipient` edge so the
+  head always lands in the graph.
+
+## P0 limitations
+
+- Coin v1 events skipped. Every in-scope token is FA v2 — USDCx, the LayerZero
+  `.e` family, and native MOVE all live as fungible assets, not legacy `Coin<T>`.
+  Coin v1 on Movement is dominated by gas fees and legacy test tokens, none of
+  which we want in the graph. Revisit if a future bridge starts minting into
+  `Coin<T>`.
+- `evm_source` is always NULL for now — neither USDCx `Mint` nor LayerZero
+  `OftReceived` carries it in the event; it's in entry-function args / LZ packet.
+  P1.
+- Reputation propagation is naive `max(current, decay * sender_score)` — cycle-
+  safe and idempotent, but not amount-weighted.
+
+## Test
+
+```
+cargo test -p processor --test address_reputation_smoke -- --nocapture
+```
+
+Builds minimal protos from real testnet txns 158025629 (USDCx mint) and
+163802127 (user FA transfer); asserts bridge head edges, owner-keyed user
+edges, and `asset_type` resolution.
+
+## Run on testnet
+
+```
+diesel migration run --database-url postgres://...
+cargo run -p processor --release -- \
+  --config processor/example-configs/address_reputation.testnet.yaml
+```

processor/src/processors/address_reputation/address_reputation_config.rs

@@ -47,6 +47,7 @@ impl BridgeConfig {
     fn default_amount_path() -> String {
         "amount".to_string()
     }
+
     const fn default_enabled() -> bool {
         true
     }
@@ -56,6 +57,7 @@ impl AddressReputationConfig {
     pub const fn default_channel_size() -> usize {
         10
     }
+
     pub const fn default_decay() -> f64 {
         0.8
     }

processor/src/processors/address_reputation/address_reputation_extractor.rs

@@ -119,8 +119,10 @@ impl Processable for AddressReputationExtractor {
                 // Bridge check (registry-driven, exact event-type match). We do this BEFORE
                 // generic FA classification so a bridge's emitted FA Deposit (which fires
                 // alongside the Mint) doesn't get double-counted as a user transfer.
-                if let Some(entry) =
-                    self.bridge_registry.iter().find(|e| e.enabled && e.event_type == type_str)
+                if let Some(entry) = self
+                    .bridge_registry
+                    .iter()
+                    .find(|e| e.enabled && e.event_type == type_str)
                 {
                     if let Some(inflow) = parse_bridge_event(
                         event.data.as_str(),
@@ -161,7 +163,11 @@ impl Processable for AddressReputationExtractor {
                     .get(&storage_id)
                     .cloned()
                     .unwrap_or_else(|| storage_id.clone());
-                let amount = match v.get("amount").and_then(|x| x.as_str()).and_then(|s| BigDecimal::from_str(s).ok()) {
+                let amount = match v
+                    .get("amount")
+                    .and_then(|x| x.as_str())
+                    .and_then(|s| BigDecimal::from_str(s).ok())
+                {
                     Some(a) => a,
                     None => continue,
                 };

processor/src/processors/address_reputation/address_reputation_storer.rs

@@ -218,7 +218,9 @@ async fn upsert_address_reputation(
         .on_conflict(ar::address)
         .do_update()
         .set((
-            ar::score.eq(sql::<Numeric>("GREATEST(address_reputation.score, EXCLUDED.score)")),
+            ar::score.eq(sql::<Numeric>(
+                "GREATEST(address_reputation.score, EXCLUDED.score)",
+            )),
             ar::highest_seed.eq(sql::<Numeric>(
                 "GREATEST(address_reputation.highest_seed, EXCLUDED.highest_seed)",
             )),

processor/tests/address_reputation_smoke.rs

@@ -116,7 +116,10 @@ fn make_mint_txn() -> Transaction {
         r#"{{"amount":"{MINT_AMOUNT}","relayer":"0xdb8069db67708d47796f837e9862ca9aae6ed5a522bc0b9b7bc25584e77577bb","recipient":"{MINT_RECIPIENT_OWNER}","fee_amount":"0","remote_token":"0x63f169ba69623ba6ccf34620857644feb46d0f87e1d7bbcf8c071d30c3d94bd6","remote_domain":10005}}"#
     );
     Transaction {
-        timestamp: Some(Timestamp { seconds: 1_700_000_000, nanos: 0 }),
+        timestamp: Some(Timestamp {
+            seconds: 1_700_000_000,
+            nanos: 0,
+        }),
         version: 158_025_629,
         info: Some(TransactionInfo {
             hash: vec![],
@@ -147,12 +150,13 @@ fn make_mint_txn() -> Transaction {
 }
 
 fn make_transfer_txn() -> Transaction {
-    let withdraw_data =
-        format!(r#"{{"store":"{SENDER_STORE}","amount":"{TRANSFER_AMOUNT}"}}"#);
-    let deposit_data =
-        format!(r#"{{"store":"{RECIPIENT_STORE}","amount":"{TRANSFER_AMOUNT}"}}"#);
+    let withdraw_data = format!(r#"{{"store":"{SENDER_STORE}","amount":"{TRANSFER_AMOUNT}"}}"#);
+    let deposit_data = format!(r#"{{"store":"{RECIPIENT_STORE}","amount":"{TRANSFER_AMOUNT}"}}"#);
     Transaction {
-        timestamp: Some(Timestamp { seconds: 1_700_000_500, nanos: 0 }),
+        timestamp: Some(Timestamp {
+            seconds: 1_700_000_500,
+            nanos: 0,
+        }),
         version: 163_802_127,
         info: Some(TransactionInfo {
             hash: vec![],
@@ -241,14 +245,25 @@ async fn extractor_emits_bridge_head_and_owner_keyed_transfer() {
     }
 
     // --- assertions ---
-    assert_eq!(inflows.len(), 1, "expected one bridge inflow for the mint txn");
+    assert_eq!(
+        inflows.len(),
+        1,
+        "expected one bridge inflow for the mint txn"
+    );
     let bi = &inflows[0];
     assert_eq!(bi.aptos_recipient, MINT_RECIPIENT_OWNER);
     assert_eq!(bi.bridge_name, "circle_usdcx");
     assert_eq!(bi.src_chain_id, Some(10005));
-    assert!(bi.evm_source.is_none(), "Mint event carries no EVM source field");
+    assert!(
+        bi.evm_source.is_none(),
+        "Mint event carries no EVM source field"
+    );
 
-    assert_eq!(edges.len(), 2, "expected one synthetic bridge edge + one user transfer edge");
+    assert_eq!(
+        edges.len(),
+        2,
+        "expected one synthetic bridge edge + one user transfer edge"
+    );
 
     let bridge_edge = edges
         .iter()