cala_ledger/entry/
repo.rs

1use crate::primitives::{AccountId, AccountSetId, DataSourceId, EntryId, JournalId, TransactionId};
2use es_entity::*;
3use sqlx::PgPool;
4
5use super::{entity::*, error::*};
6
7#[derive(EsRepo, Debug, Clone)]
8#[es_repo(
9    entity = "Entry",
10    err = "EntryError",
11    columns(
12        account_id(ty = "AccountId", list_for, update(persist = false)),
13        journal_id(ty = "JournalId", list_for, update(persist = false)),
14        transaction_id(ty = "TransactionId", update(persist = false)),
15        data_source_id(
16            ty = "DataSourceId",
17            create(accessor = "data_source().into()"),
18            update(persist = false),
19        ),
20    ),
21    tbl_prefix = "cala"
22)]
23pub(crate) struct EntryRepo {
24    #[allow(dead_code)]
25    pool: PgPool,
26}
27
28impl EntryRepo {
29    pub(crate) fn new(pool: &PgPool) -> Self {
30        Self { pool: pool.clone() }
31    }
32
33    pub(super) async fn list_for_account_set_id_by_created_at(
34        &self,
35        account_set_id: AccountSetId,
36        query: es_entity::PaginatedQueryArgs<entry_cursor::EntriesByCreatedAtCursor>,
37        direction: es_entity::ListDirection,
38    ) -> Result<
39        es_entity::PaginatedQueryRet<Entry, entry_cursor::EntriesByCreatedAtCursor>,
40        EntryError,
41    > {
42        let es_entity::PaginatedQueryArgs { first, after } = query;
43        let (id, created_at) = if let Some(after) = after {
44            (Some(after.id), Some(after.created_at))
45        } else {
46            (None, None)
47        };
48
49        let executor = &self.pool;
50
51        let (entities, has_next_page) = match direction {
52                    es_entity::ListDirection::Ascending => {
53                        es_entity::es_query!(
54                            "cala",
55                            executor,
56                            r#"
57                            SELECT created_at, id
58                            FROM cala_entries
59                            JOIN cala_balance_history ON cala_entries.id = cala_balance_history.latest_entry_id
60                            WHERE cala_balance_history.account_id = $4
61                              AND (COALESCE((created_at, id) > ($3, $2), $2 IS NULL))
62                            ORDER BY created_at ASC, id ASC
63                            LIMIT $1"#,
64                            (first + 1) as i64,
65                            id as Option<EntryId>,
66                            created_at as Option<chrono::DateTime<chrono::Utc>>,
67                            account_set_id as AccountSetId,
68                        )
69                            .fetch_n(first)
70                            .await?
71                    },
72                    es_entity::ListDirection::Descending => {
73                        es_entity::es_query!(
74                            "cala",
75                            executor,
76                            r#"
77                            SELECT created_at, id
78                            FROM cala_entries
79                            JOIN cala_balance_history ON cala_entries.id = cala_balance_history.latest_entry_id
80                            WHERE cala_balance_history.account_id = $4
81                              AND (COALESCE((created_at, id) < ($3, $2), $2 IS NULL))
82                            ORDER BY created_at DESC, id DESC
83                            LIMIT $1"#,
84                            (first + 1) as i64,
85                            id as Option<EntryId>,
86                            created_at as Option<chrono::DateTime<chrono::Utc>>,
87                            account_set_id as AccountSetId,
88                        )
89                            .fetch_n(first)
90                            .await?
91                    },
92                };
93
94        let end_cursor = entities
95            .last()
96            .map(entry_cursor::EntriesByCreatedAtCursor::from);
97
98        Ok(es_entity::PaginatedQueryRet {
99            entities,
100            has_next_page,
101            end_cursor,
102        })
103    }
104
105    #[cfg(feature = "import")]
106    pub(super) async fn import(
107        &self,
108        op: &mut DbOp<'_>,
109        origin: DataSourceId,
110        entry: &mut Entry,
111    ) -> Result<(), EntryError> {
112        let recorded_at = op.now();
113        sqlx::query!(
114            r#"INSERT INTO cala_entries (data_source_id, id, journal_id, account_id, created_at)
115            VALUES ($1, $2, $3, $4, $5)"#,
116            origin as DataSourceId,
117            entry.values().id as EntryId,
118            entry.values().journal_id as JournalId,
119            entry.values().account_id as AccountId,
120            recorded_at,
121        )
122        .execute(&mut **op.tx())
123        .await?;
124        self.persist_events(op, &mut entry.events).await?;
125        Ok(())
126    }
127}
cala_ledger/entry/repo.rs

cala_ledger/entry/
repo.rs
