chore: account set member

cala-ledger-core-types/src/account_set.rs

@@ -15,19 +15,19 @@ pub struct AccountSetValues {
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(tag = "type", content = "id")]
-pub enum AccountSetMember {
+pub enum AccountSetMemberId {
     Account(AccountId),
     AccountSet(AccountSetId),
 }
 
-impl From<AccountId> for AccountSetMember {
+impl From<AccountId> for AccountSetMemberId {
     fn from(account_id: AccountId) -> Self {
-        AccountSetMember::Account(account_id)
+        Self::Account(account_id)
     }
 }
 
-impl From<AccountSetId> for AccountSetMember {
+impl From<AccountSetId> for AccountSetMemberId {
     fn from(id: AccountSetId) -> Self {
-        AccountSetMember::AccountSet(id)
+        Self::AccountSet(id)
     }
 }

cala-ledger-core-types/src/outbox.rs

@@ -49,12 +49,12 @@ pub enum OutboxEventPayload {
     AccountSetMemberCreated {
         source: DataSource,
         account_set_id: AccountSetId,
-        member: AccountSetMember,
+        member_id: AccountSetMemberId,
     },
     AccountSetMemberRemoved {
         source: DataSource,
         account_set_id: AccountSetId,
-        member: AccountSetMember,
+        member_id: AccountSetMemberId,
     },
     JournalCreated {
         source: DataSource,

cala-ledger-outbox-client/src/convert.rs

@@ -83,17 +83,17 @@ impl TryFrom<proto::cala_ledger_event::Payload> for OutboxEventPayload {
                 AccountSetMemberCreated {
                     source: data_source_id.parse()?,
                     account_set_id: member.account_set_id.parse()?,
-                    member: match member
+                    member_id: match member
                         .member
                         .ok_or(CalaLedgerOutboxClientError::MissingField)?
                     {
                         proto::account_set_member::Member::MemberAccountId(account_id) => {
-                            cala_types::account_set::AccountSetMember::from(
+                            cala_types::account_set::AccountSetMemberId::from(
                                 account_id.parse::<AccountId>()?,
                             )
                         }
                         proto::account_set_member::Member::MemberAccountSetId(account_set_id) => {
-                            cala_types::account_set::AccountSetMember::from(
+                            cala_types::account_set::AccountSetMemberId::from(
                                 account_set_id.parse::<AccountSetId>()?,
                             )
                         }
@@ -110,17 +110,17 @@ impl TryFrom<proto::cala_ledger_event::Payload> for OutboxEventPayload {
                 AccountSetMemberRemoved {
                     source: data_source_id.parse()?,
                     account_set_id: member.account_set_id.parse()?,
-                    member: match member
+                    member_id: match member
                         .member
                         .ok_or(CalaLedgerOutboxClientError::MissingField)?
                     {
                         proto::account_set_member::Member::MemberAccountId(account_id) => {
-                            cala_types::account_set::AccountSetMember::from(
+                            cala_types::account_set::AccountSetMemberId::from(
                                 account_id.parse::<AccountId>()?,
                             )
                         }
                         proto::account_set_member::Member::MemberAccountSetId(account_set_id) => {
-                            cala_types::account_set::AccountSetMember::from(
+                            cala_types::account_set::AccountSetMemberId::from(
                                 account_set_id.parse::<AccountSetId>()?,
                             )
                         }

cala-ledger/src/account/mod.rs

@@ -67,6 +67,15 @@ impl Accounts {
         self.repo.find_all(account_ids).await
     }
 
+    #[instrument(name = "cala_ledger.accounts.find_all", skip(self, op), err)]
+    pub async fn find_all_in_op<T: From<Account>>(
+        &self,
+        op: &mut AtomicOperation<'_>,
+        account_ids: &[AccountId],
+    ) -> Result<HashMap<AccountId, T>, AccountError> {
+        self.repo.find_all_in_tx(op.tx(), account_ids).await
+    }
+
     #[instrument(name = "cala_ledger.accounts.find_by_external_id", skip(self), err)]
     pub async fn find_by_external_id(&self, external_id: String) -> Result<Account, AccountError> {
         self.repo.find_by_external_id(external_id).await

cala-ledger/src/account/repo.rs

@@ -1,6 +1,6 @@
 #[cfg(feature = "import")]
 use chrono::{DateTime, Utc};
-use sqlx::{PgPool, Postgres, Transaction};
+use sqlx::{Executor, PgPool, Postgres, Transaction};
 
 use std::collections::HashMap;
 
@@ -92,6 +92,22 @@ impl AccountRepo {
     pub(super) async fn find_all<T: From<Account>>(
         &self,
         ids: &[AccountId],
+    ) -> Result<HashMap<AccountId, T>, AccountError> {
+        self.find_all_in_executor(&self.pool, ids).await
+    }
+
+    pub(super) async fn find_all_in_tx<T: From<Account>>(
+        &self,
+        tx: &mut Transaction<'_, Postgres>,
+        ids: &[AccountId],
+    ) -> Result<HashMap<AccountId, T>, AccountError> {
+        self.find_all_in_executor(&mut **tx, ids).await
+    }
+
+    async fn find_all_in_executor<T: From<Account>>(
+        &self,
+        executor: impl Executor<'_, Database = Postgres>,
+        ids: &[AccountId],
     ) -> Result<HashMap<AccountId, T>, AccountError> {
         let rows = sqlx::query_as!(
             GenericEvent,
@@ -106,7 +122,7 @@ impl AccountRepo {
             ORDER BY a.id, e.sequence"#,
             ids as &[AccountId]
         )
-        .fetch_all(&self.pool)
+        .fetch_all(executor)
         .await?;
         let n = rows.len();
         let ret = EntityEvents::load_n(rows, n)?

cala-ledger/src/account_set/cursor.rs

@@ -15,3 +15,8 @@ impl From<&AccountSetValues> for AccountSetByNameCursor {
         }
     }
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AccountSetMemberCursor {
+    pub member_created_at: chrono::DateTime<chrono::Utc>,
+}