Paulo/upstream changes 1.9.1 (#4)

* chore: bump Pallas to v0.30.1 (txpipe#811)

* Release v1.9.1

* chore: bump version

* fix: pubsub emulator again

---------

Co-authored-by: Santiago Carmuega <santiago@carmuega.me>

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "oura"
 description = "The tail of Cardano"
-version = "1.10.0"
+version = "1.10.1"
 edition = "2021"
 repository = "https://github.com/txpipe/oura"
 homepage = "https://github.com/txpipe/oura"
@@ -14,11 +14,11 @@ authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 [dependencies]
 pallas-multiplexer = "0.18.2"
 pallas-miniprotocols = "0.18.2"
-pallas-primitives = "0.29.0"
-pallas-traverse = "0.29.0"
-pallas-addresses = "0.29.0"
-pallas-codec = "0.29.0"
-pallas-crypto = "0.29.0"
+pallas-primitives = "0.30.1"
+pallas-traverse = "0.30.1"
+pallas-addresses = "0.30.1"
+pallas-codec = "0.30.1"
+pallas-crypto = "0.30.1"
 # pallas = { git = "https://github.com/txpipe/pallas" }
 # pallas = { path = "../pallas/pallas" }
 hex = "0.4.3"

src/mapper/conway.rs

@@ -1,17 +1,19 @@
-use pallas_codec::utils::{KeepRaw, NonZeroInt};
+use pallas_codec::utils::{KeepRaw, NonEmptyKeyValuePairs, NonZeroInt};
 
 use pallas_primitives::conway::{
-    AuxiliaryData, Certificate, MintedBlock, MintedDatumOption, MintedPostAlonzoTransactionOutput,
-    MintedTransactionBody, MintedTransactionOutput, MintedWitnessSet, Multiasset, NetworkId,
-    RedeemerTag, RedeemersKey, RedeemersValue,
+    AuxiliaryData, Certificate, Coin, MintedBlock, MintedDatumOption,
+    MintedPostAlonzoTransactionOutput, MintedTransactionBody, MintedTransactionOutput,
+    MintedWitnessSet, Multiasset, NetworkId, RedeemerTag, RedeemersKey, RedeemersValue,
+    RewardAccount, Value,
 };
 
 use pallas_crypto::hash::Hash;
 use pallas_primitives::ToCanonicalJson as _;
 use pallas_traverse::OriginalHash;
 
 use crate::model::{
-    BlockRecord, Era, MintRecord, PlutusRedeemerRecord, TransactionRecord, TxOutputRecord,
+    BlockRecord, Era, MintRecord, OutputAssetRecord, PlutusRedeemerRecord, TransactionRecord,
+    TxOutputRecord, WithdrawalRecord,
 };
 use crate::utils::time::TimeProvider;
 use crate::{
@@ -22,6 +24,27 @@ use crate::{
 use super::{map::ToHex, EventWriter};
 
 impl EventWriter {
+    pub fn collect_conway_coin_value(&self, amount: &Value) -> u64 {
+        match amount {
+            Value::Coin(x) => *x,
+            Value::Multiasset(x, _) => *x,
+        }
+    }
+
+    pub fn collect_conway_asset_records(&self, amount: &Value) -> Vec<OutputAssetRecord> {
+        match amount {
+            Value::Coin(_) => vec![],
+            Value::Multiasset(_, policies) => policies
+                .iter()
+                .flat_map(|(policy, assets)| {
+                    assets.iter().map(|(asset, amount)| {
+                        self.to_transaction_output_asset_record(policy, asset, u64::from(amount))
+                    })
+                })
+                .collect(),
+        }
+    }
+
     pub fn collect_conway_mint_records(&self, mint: &Multiasset<NonZeroInt>) -> Vec<MintRecord> {
         mint.iter()
             .flat_map(|(policy, assets)| {
@@ -50,8 +73,8 @@ impl EventWriter {
 
         Ok(TxOutputRecord {
             address: address.to_string(),
-            amount: super::map::get_tx_output_coin_value(&output.value),
-            assets: self.collect_asset_records(&output.value).into(),
+            amount: self.collect_conway_coin_value(&output.value),
+            assets: self.collect_conway_asset_records(&output.value).into(),
             datum_hash: match &output.datum_option {
                 Some(MintedDatumOption::Hash(x)) => Some(x.to_string()),
                 Some(MintedDatumOption::Data(x)) => Some(x.original_hash().to_hex()),
@@ -98,6 +121,22 @@ impl EventWriter {
             .collect()
     }
 
+    pub fn collect_conway_withdrawal_records(
+        &self,
+        withdrawls: &NonEmptyKeyValuePairs<RewardAccount, Coin>,
+    ) -> Vec<WithdrawalRecord> {
+        withdrawls
+            .iter()
+            .map(|(reward_account, coin)| WithdrawalRecord {
+                reward_account: {
+                    let hex = reward_account.to_hex();
+                    hex.strip_prefix("e1").map(|x| x.to_string()).unwrap_or(hex)
+                },
+                coin: *coin,
+            })
+            .collect()
+    }
+
     pub fn to_conway_tx_size(
         &self,
         body: &KeepRaw<MintedTransactionBody>,
@@ -243,7 +282,7 @@ impl EventWriter {
             }
 
             if let Some(withdrawals) = &body.withdrawals {
-                record.withdrawals = self.collect_withdrawal_records(withdrawals).into();
+                record.withdrawals = self.collect_conway_withdrawal_records(withdrawals).into();
             }
         }
 
@@ -317,6 +356,22 @@ impl EventWriter {
             .collect()
     }
 
+    pub fn crawl_conway_transaction_output_amount(&self, amount: &Value) -> Result<(), Error> {
+        if let Value::Multiasset(_, policies) = amount {
+            for (policy, assets) in policies.iter() {
+                for (asset, amount) in assets.iter() {
+                    self.append_from(self.to_transaction_output_asset_record(
+                        policy,
+                        asset,
+                        u64::from(amount),
+                    ))?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     fn crawl_conway_output(&self, output: &MintedPostAlonzoTransactionOutput) -> Result<(), Error> {
         let record = self.to_conway_output_record(output)?;
         self.append(record.into())?;
@@ -328,7 +383,7 @@ impl EventWriter {
             ..EventContext::default()
         });
 
-        child.crawl_transaction_output_amount(&output.value)?;
+        child.crawl_conway_transaction_output_amount(&output.value)?;
 
         if let Some(MintedDatumOption::Data(datum)) = &output.datum_option {
             let record = self.to_plutus_datum_record(datum)?;

src/sinks/gcp_pubsub/run.rs

@@ -57,16 +57,7 @@ pub fn writer_loop(
         .build()?;
 
     let publisher: Publisher = rt.block_on(async {
-        let client_config = if emulator {
-            ClientConfig {
-                project_id: Some(emulator_project_id.clone().unwrap_or_default()),
-                environment: Environment::Emulator(emulator_endpoint.clone().unwrap_or_default()),
-                ..Default::default()
-            }
-        } else {
-            ClientConfig::default()
-        };
-        let client = Client::new(client_config).await?;
+        let client = Client::new(ClientConfig::default().with_auth().await?).await?;
         let topic = client.topic(topic_name);
         Result::<_, crate::Error>::Ok(topic.new_publisher(None))
     })?;

src/sinks/gcp_pubsub/setup.rs

@@ -16,10 +16,6 @@ pub struct Config {
     pub retry_policy: Option<retry::Policy>,
     pub ordering_key: Option<String>,
     pub attributes: Option<GenericKV>,
-    pub emulator: Option<bool>,
-    pub emulator_endpoint: Option<String>,
-    pub emulator_project_id: Option<String>,
-
     #[warn(deprecated)]
     pub credentials: Option<String>,
 }