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", list_for, 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                            entity = Entry,
55                            r#"
56                            SELECT created_at, id
57                            FROM cala_entries
58                            JOIN cala_balance_history ON cala_entries.id = cala_balance_history.latest_entry_id
59                            WHERE cala_balance_history.account_id = $4
60                              AND (COALESCE((created_at, id) > ($3, $2), $2 IS NULL))
61                            ORDER BY created_at ASC, id ASC
62                            LIMIT $1"#,
63                            (first + 1) as i64,
64                            id as Option<EntryId>,
65                            created_at as Option<chrono::DateTime<chrono::Utc>>,
66                            account_set_id as AccountSetId,
67                        )
68                            .fetch_n(executor, first)
69                            .await?
70                    },
71                    es_entity::ListDirection::Descending => {
72                        es_entity::es_query!(
73                            entity = Entry,
74                            r#"
75                            SELECT created_at, id
76                            FROM cala_entries
77                            JOIN cala_balance_history ON cala_entries.id = cala_balance_history.latest_entry_id
78                            WHERE cala_balance_history.account_id = $4
79                              AND (COALESCE((created_at, id) < ($3, $2), $2 IS NULL))
80                            ORDER BY created_at DESC, id DESC
81                            LIMIT $1"#,
82                            (first + 1) as i64,
83                            id as Option<EntryId>,
84                            created_at as Option<chrono::DateTime<chrono::Utc>>,
85                            account_set_id as AccountSetId,
86                        )
87                            .fetch_n(executor, first)
88                            .await?
89                    },
90                };
91
92        let end_cursor = entities
93            .last()
94            .map(entry_cursor::EntriesByCreatedAtCursor::from);
95
96        Ok(es_entity::PaginatedQueryRet {
97            entities,
98            has_next_page,
99            end_cursor,
100        })
101    }
102
103    #[cfg(feature = "import")]
104    pub(super) async fn import(
105        &self,
106        op: &mut impl es_entity::AtomicOperation,
107        origin: DataSourceId,
108        entry: &mut Entry,
109    ) -> Result<(), EntryError> {
110        let recorded_at = op.now();
111        sqlx::query!(
112            r#"INSERT INTO cala_entries (data_source_id, id, journal_id, account_id, created_at)
113            VALUES ($1, $2, $3, $4, $5)"#,
114            origin as DataSourceId,
115            entry.values().id as EntryId,
116            entry.values().journal_id as JournalId,
117            entry.values().account_id as AccountId,
118            recorded_at,
119        )
120        .execute(op.as_executor())
121        .await?;
122        self.persist_events(op, entry.events_mut()).await?;
123        Ok(())
124    }
125}
cala_ledger/entry/repo.rs

cala_ledger/entry/
repo.rs
