1use async_graphql::{dataloader::*, types::connection::*, *};
2use cala_ledger::{balance::AccountBalance, primitives::*, tx_template::NewParamDefinition};
3
4use crate::{app::CalaApp, extension::*};
5
6use super::{
7 account::*, account_set::*, balance::*, journal::*, loader::*, primitives::*, transaction::*,
8 tx_template::*, velocity::*,
9};
10
11#[derive(Default)]
12pub struct CoreQuery<E: QueryExtensionMarker> {
13 _phantom: std::marker::PhantomData<E>,
14}
15
16#[Object(name = "Query")]
17impl<E: QueryExtensionMarker> CoreQuery<E> {
18 #[graphql(flatten)]
19 async fn extension(&self) -> E {
20 E::default()
21 }
22
23 async fn account(&self, ctx: &Context<'_>, id: UUID) -> async_graphql::Result<Option<Account>> {
24 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
25 Ok(loader.load_one(AccountId::from(id)).await?)
26 }
27
28 async fn account_by_external_id(
29 &self,
30 ctx: &Context<'_>,
31 external_id: String,
32 ) -> async_graphql::Result<Option<Account>> {
33 let app = ctx.data_unchecked::<CalaApp>();
34 match app
35 .ledger()
36 .accounts()
37 .find_by_external_id(external_id)
38 .await
39 {
40 Ok(account) => Ok(Some(account.into())),
41 Err(cala_ledger::account::error::AccountError::CouldNotFindByExternalId(_)) => Ok(None),
42 Err(err) => Err(err.into()),
43 }
44 }
45
46 async fn account_by_code(
47 &self,
48 ctx: &Context<'_>,
49 code: String,
50 ) -> async_graphql::Result<Option<Account>> {
51 let app = ctx.data_unchecked::<CalaApp>();
52 match app.ledger().accounts().find_by_code(code).await {
53 Ok(account) => Ok(Some(account.into())),
54 Err(cala_ledger::account::error::AccountError::CouldNotFindByCode(_)) => Ok(None),
55 Err(err) => Err(err.into()),
56 }
57 }
58
59 async fn accounts(
60 &self,
61 ctx: &Context<'_>,
62 first: i32,
63 after: Option<String>,
64 ) -> Result<Connection<AccountsByNameCursor, Account, EmptyFields, EmptyFields>> {
65 let app = ctx.data_unchecked::<CalaApp>();
66 query(
67 after,
68 None,
69 Some(first),
70 None,
71 |after, _, first, _| async move {
72 let first = first.expect("First always exists");
73 let result = app
74 .ledger()
75 .accounts()
76 .list(cala_ledger::es_entity::PaginatedQueryArgs { first, after })
77 .await?;
78 let mut connection = Connection::new(false, result.has_next_page);
79 connection
80 .edges
81 .extend(result.entities.into_iter().map(|entity| {
82 let cursor = AccountsByNameCursor::from(&entity);
83 Edge::new(cursor, Account::from(entity))
84 }));
85 Ok::<_, async_graphql::Error>(connection)
86 },
87 )
88 .await
89 }
90
91 async fn account_set(
92 &self,
93 ctx: &Context<'_>,
94 id: UUID,
95 ) -> async_graphql::Result<Option<AccountSet>> {
96 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
97 Ok(loader.load_one(AccountSetId::from(id)).await?)
98 }
99
100 async fn journal(&self, ctx: &Context<'_>, id: UUID) -> async_graphql::Result<Option<Journal>> {
101 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
102 Ok(loader.load_one(JournalId::from(id)).await?)
103 }
104
105 async fn balance(
106 &self,
107 ctx: &Context<'_>,
108 journal_id: UUID,
109 account_id: UUID,
110 currency: CurrencyCode,
111 ) -> async_graphql::Result<Option<Balance>> {
112 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
113 let journal_id = JournalId::from(journal_id);
114 let account_id = AccountId::from(account_id);
115 let currency = Currency::from(currency);
116 let balance: Option<AccountBalance> =
117 loader.load_one((journal_id, account_id, currency)).await?;
118 Ok(balance.map(Balance::from))
119 }
120
121 async fn transaction(
122 &self,
123 ctx: &Context<'_>,
124 id: UUID,
125 ) -> async_graphql::Result<Option<Transaction>> {
126 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
127 Ok(loader.load_one(TransactionId::from(id)).await?)
128 }
129
130 async fn transaction_by_external_id(
131 &self,
132 ctx: &Context<'_>,
133 external_id: String,
134 ) -> async_graphql::Result<Option<Transaction>> {
135 let app = ctx.data_unchecked::<CalaApp>();
136 match app
137 .ledger()
138 .transactions()
139 .find_by_external_id(external_id)
140 .await
141 {
142 Ok(transaction) => Ok(Some(transaction.into())),
143 Err(cala_ledger::transaction::error::TransactionError::CouldNotFindByExternalId(_)) => {
144 Ok(None)
145 }
146 Err(err) => Err(err.into()),
147 }
148 }
149
150 async fn tx_template(
151 &self,
152 ctx: &Context<'_>,
153 id: UUID,
154 ) -> async_graphql::Result<Option<TxTemplate>> {
155 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
156 Ok(loader.load_one(TxTemplateId::from(id)).await?)
157 }
158
159 async fn tx_template_by_code(
160 &self,
161 ctx: &Context<'_>,
162 code: String,
163 ) -> async_graphql::Result<Option<TxTemplate>> {
164 let app = ctx.data_unchecked::<CalaApp>();
165 match app.ledger().tx_templates().find_by_code(code).await {
166 Ok(tx_template) => Ok(Some(tx_template.into())),
167 Err(cala_ledger::tx_template::error::TxTemplateError::CouldNotFindByCode(_)) => {
168 Ok(None)
169 }
170 Err(err) => Err(err.into()),
171 }
172 }
173
174 async fn velocity_limit(
175 &self,
176 ctx: &Context<'_>,
177 id: UUID,
178 ) -> async_graphql::Result<Option<VelocityLimit>> {
179 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
180 Ok(loader.load_one(VelocityLimitId::from(id)).await?)
181 }
182
183 async fn velocity_control(
184 &self,
185 ctx: &Context<'_>,
186 id: UUID,
187 ) -> async_graphql::Result<Option<VelocityControl>> {
188 let loader = ctx.data_unchecked::<DataLoader<LedgerDataLoader>>();
189 Ok(loader.load_one(VelocityControlId::from(id)).await?)
190 }
191}
192
193#[derive(Default)]
194pub struct CoreMutation<E: MutationExtensionMarker> {
195 _phantom: std::marker::PhantomData<E>,
196}
197
198#[Object(name = "Mutation")]
199impl<E: MutationExtensionMarker> CoreMutation<E> {
200 #[graphql(flatten)]
201 async fn extension(&self) -> E {
202 E::default()
203 }
204
205 async fn account_create(
206 &self,
207 ctx: &Context<'_>,
208 input: AccountCreateInput,
209 ) -> Result<AccountCreatePayload> {
210 let app = ctx.data_unchecked::<CalaApp>();
211 let mut op = ctx
212 .data_unchecked::<DbOp>()
213 .try_lock()
214 .expect("Lock held concurrently");
215 let mut builder = cala_ledger::account::NewAccount::builder();
216 builder
217 .id(input.account_id)
218 .name(input.name)
219 .code(input.code)
220 .normal_balance_type(input.normal_balance_type)
221 .status(input.status);
222
223 if let Some(external_id) = input.external_id {
224 builder.external_id(external_id);
225 }
226 if let Some(description) = input.description {
227 builder.description(description);
228 }
229 if let Some(metadata) = input.metadata {
230 builder.metadata(metadata)?;
231 }
232 let account = app
233 .ledger()
234 .accounts()
235 .create_in_op(&mut op, builder.build()?)
236 .await?;
237
238 if let Some(account_set_ids) = input.account_set_ids {
239 for id in account_set_ids {
240 app.ledger()
241 .account_sets()
242 .add_member_in_op(&mut op, AccountSetId::from(id), account.id())
243 .await?;
244 }
245 }
246
247 Ok(account.into())
248 }
249
250 async fn account_update(
251 &self,
252 ctx: &Context<'_>,
253 id: UUID,
254 input: AccountUpdateInput,
255 ) -> Result<AccountUpdatePayload> {
256 let app = ctx.data_unchecked::<CalaApp>();
257 let mut op = ctx
258 .data_unchecked::<DbOp>()
259 .try_lock()
260 .expect("Lock held concurrently");
261
262 let mut builder = cala_ledger::account::AccountUpdate::default();
263 if let Some(name) = input.name {
264 builder.name(name);
265 }
266 if let Some(code) = input.code {
267 builder.code(code);
268 }
269 if let Some(normal_balance_type) = input.normal_balance_type {
270 builder.normal_balance_type(normal_balance_type);
271 }
272 if let Some(status) = input.status {
273 builder.status(status);
274 }
275 if let Some(external_id) = input.external_id {
276 builder.external_id(external_id);
277 }
278 if let Some(description) = input.description {
279 builder.description(description);
280 }
281 if let Some(metadata) = input.metadata {
282 builder.metadata(metadata)?;
283 }
284
285 let mut account = app.ledger().accounts().find(AccountId::from(id)).await?;
286 account.update(builder);
287 app.ledger()
288 .accounts()
289 .persist_in_op(&mut op, &mut account)
290 .await?;
291
292 Ok(account.into())
293 }
294
295 async fn account_set_create(
296 &self,
297 ctx: &Context<'_>,
298 input: AccountSetCreateInput,
299 ) -> Result<AccountSetCreatePayload> {
300 let app = ctx.data_unchecked::<CalaApp>();
301 let mut op = ctx
302 .data_unchecked::<DbOp>()
303 .try_lock()
304 .expect("Lock held concurrently");
305 let mut builder = cala_ledger::account_set::NewAccountSet::builder();
306 builder
307 .id(input.account_set_id)
308 .journal_id(input.journal_id)
309 .name(input.name)
310 .normal_balance_type(input.normal_balance_type);
311
312 if let Some(description) = input.description {
313 builder.description(description);
314 }
315 if let Some(metadata) = input.metadata {
316 builder.metadata(metadata)?;
317 }
318 let account_set = app
319 .ledger()
320 .account_sets()
321 .create_in_op(&mut op, builder.build()?)
322 .await?;
323
324 Ok(account_set.into())
325 }
326
327 async fn account_set_update(
328 &self,
329 ctx: &Context<'_>,
330 id: UUID,
331 input: AccountSetUpdateInput,
332 ) -> Result<AccountSetUpdatePayload> {
333 let app = ctx.data_unchecked::<CalaApp>();
334 let mut op = ctx
335 .data_unchecked::<DbOp>()
336 .try_lock()
337 .expect("Lock held concurrently");
338 let mut builder = cala_ledger::account_set::AccountSetUpdate::default();
339 if let Some(name) = input.name {
340 builder.name(name);
341 }
342 if let Some(desc) = input.description {
343 builder.description(desc);
344 }
345 if let Some(normal_balance_type) = input.normal_balance_type {
346 builder.normal_balance_type(normal_balance_type);
347 }
348
349 if let Some(metadata) = input.metadata {
350 builder.metadata(metadata)?;
351 }
352
353 let mut account_set = app
354 .ledger()
355 .account_sets()
356 .find(AccountSetId::from(id))
357 .await?;
358 account_set.update(builder);
359
360 app.ledger()
361 .account_sets()
362 .persist_in_op(&mut op, &mut account_set)
363 .await?;
364
365 Ok(account_set.into())
366 }
367
368 async fn add_to_account_set(
369 &self,
370 ctx: &Context<'_>,
371 input: AddToAccountSetInput,
372 ) -> Result<AddToAccountSetPayload> {
373 let app = ctx.data_unchecked::<CalaApp>();
374 let mut op = ctx
375 .data_unchecked::<DbOp>()
376 .try_lock()
377 .expect("Lock held concurrently");
378
379 let account_set = app
380 .ledger()
381 .account_sets()
382 .add_member_in_op(&mut op, AccountSetId::from(input.account_set_id), input)
383 .await?;
384
385 Ok(account_set.into())
386 }
387
388 async fn remove_from_account_set(
389 &self,
390 ctx: &Context<'_>,
391 input: RemoveFromAccountSetInput,
392 ) -> Result<RemoveFromAccountSetPayload> {
393 let app = ctx.data_unchecked::<CalaApp>();
394 let mut op = ctx
395 .data_unchecked::<DbOp>()
396 .try_lock()
397 .expect("Lock held concurrently");
398
399 let account_set = app
400 .ledger()
401 .account_sets()
402 .remove_member_in_op(&mut op, AccountSetId::from(input.account_set_id), input)
403 .await?;
404
405 Ok(account_set.into())
406 }
407
408 async fn journal_create(
409 &self,
410 ctx: &Context<'_>,
411 input: JournalCreateInput,
412 ) -> Result<JournalCreatePayload> {
413 let app = ctx.data_unchecked::<CalaApp>();
414 let mut op = ctx
415 .data_unchecked::<DbOp>()
416 .try_lock()
417 .expect("Lock held concurrently");
418 let mut builder = cala_ledger::journal::NewJournal::builder();
419 builder
420 .id(input.journal_id)
421 .name(input.name)
422 .status(input.status);
423 if let Some(description) = input.description {
424 builder.description(description);
425 }
426 let journal = app
427 .ledger()
428 .journals()
429 .create_in_op(&mut op, builder.build()?)
430 .await?;
431
432 Ok(journal.into())
433 }
434
435 async fn journal_update(
436 &self,
437 ctx: &Context<'_>,
438 id: UUID,
439 input: JournalUpdateInput,
440 ) -> Result<JournalUpdatePayload> {
441 let app = ctx.data_unchecked::<CalaApp>();
442 let mut op = ctx
443 .data_unchecked::<DbOp>()
444 .try_lock()
445 .expect("Lock held concurrently");
446
447 let mut builder = cala_ledger::journal::JournalUpdate::default();
448 if let Some(name) = input.name {
449 builder.name(name);
450 }
451 if let Some(status) = input.status {
452 builder.status(status);
453 }
454 if let Some(description) = input.description {
455 builder.description(description);
456 }
457
458 let mut journal = app.ledger().journals().find(JournalId::from(id)).await?;
459 journal.update(builder);
460
461 app.ledger()
462 .journals()
463 .persist_in_op(&mut op, &mut journal)
464 .await?;
465
466 Ok(journal.into())
467 }
468
469 async fn tx_template_create(
470 &self,
471 ctx: &Context<'_>,
472 input: TxTemplateCreateInput,
473 ) -> Result<TxTemplateCreatePayload> {
474 let app = ctx.data_unchecked::<CalaApp>();
475 let mut op = ctx
476 .data_unchecked::<DbOp>()
477 .try_lock()
478 .expect("Lock held concurrently");
479 let mut new_tx_template_transaction_builder =
480 cala_ledger::tx_template::NewTxTemplateTransaction::builder();
481 let TxTemplateTransactionInput {
482 effective,
483 journal_id,
484 correlation_id,
485 external_id,
486 description,
487 metadata,
488 } = input.transaction;
489 new_tx_template_transaction_builder
490 .effective(effective)
491 .journal_id(journal_id);
492 if let Some(correlation_id) = correlation_id {
493 new_tx_template_transaction_builder.correlation_id(correlation_id);
494 };
495 if let Some(external_id) = external_id {
496 new_tx_template_transaction_builder.external_id(external_id);
497 };
498 if let Some(description) = description {
499 new_tx_template_transaction_builder.description(description);
500 };
501 if let Some(metadata) = metadata {
502 new_tx_template_transaction_builder.metadata(metadata);
503 }
504 let new_transaction = new_tx_template_transaction_builder.build()?;
505
506 let mut new_params = Vec::new();
507 if let Some(params) = input.params {
508 for param in params {
509 let mut param_builder = NewParamDefinition::builder();
510 param_builder.name(param.name).r#type(param.r#type.into());
511 if let Some(default) = param.default {
512 param_builder.default_expr(default);
513 }
514 if let Some(desc) = param.description {
515 param_builder.description(desc);
516 }
517 let new_param = param_builder.build()?;
518 new_params.push(new_param);
519 }
520 }
521
522 let mut new_entries = Vec::new();
523 for entry in input.entries {
524 let TxTemplateEntryInput {
525 entry_type,
526 account_id,
527 layer,
528 direction,
529 units,
530 currency,
531 description,
532 } = entry;
533 let mut new_entry_input_builder =
534 cala_ledger::tx_template::NewTxTemplateEntry::builder();
535 new_entry_input_builder
536 .entry_type(entry_type)
537 .account_id(account_id)
538 .layer(layer)
539 .direction(direction)
540 .units(units)
541 .currency(currency);
542 if let Some(desc) = description {
543 new_entry_input_builder.description(desc);
544 }
545 let new_entry_input = new_entry_input_builder.build()?;
546 new_entries.push(new_entry_input);
547 }
548
549 let mut new_tx_template_builder = cala_ledger::tx_template::NewTxTemplate::builder();
550 new_tx_template_builder
551 .id(input.tx_template_id)
552 .code(input.code)
553 .transaction(new_transaction)
554 .params(new_params)
555 .entries(new_entries);
556 if let Some(desc) = input.description {
557 new_tx_template_builder.description(desc);
558 }
559 if let Some(metadata) = input.metadata {
560 new_tx_template_builder.metadata(metadata)?;
561 }
562 let new_tx_template = new_tx_template_builder.build()?;
563
564 let tx_template = app
565 .ledger()
566 .tx_templates()
567 .create_in_op(&mut op, new_tx_template)
568 .await?;
569
570 Ok(tx_template.into())
571 }
572
573 async fn transaction_post(
574 &self,
575 ctx: &Context<'_>,
576 input: TransactionInput,
577 ) -> Result<TransactionPostPayload> {
578 let app = ctx.data_unchecked::<CalaApp>();
579 let mut op = ctx
580 .data_unchecked::<DbOp>()
581 .try_lock()
582 .expect("Lock held concurrently");
583 let params = input.params.map(cala_ledger::tx_template::Params::from);
584 let transaction = app
585 .ledger()
586 .post_transaction_in_op(
587 &mut op,
588 input.transaction_id.into(),
589 &input.tx_template_code,
590 params.unwrap_or_default(),
591 )
592 .await?;
593 Ok(transaction.into())
594 }
595
596 async fn velocity_limit_create(
597 &self,
598 ctx: &Context<'_>,
599 input: VelocityLimitCreateInput,
600 ) -> Result<VelocityLimitCreatePayload> {
601 let app = ctx.data_unchecked::<CalaApp>();
602 let mut op = ctx
603 .data_unchecked::<DbOp>()
604 .try_lock()
605 .expect("Lock held concurrently");
606
607 let mut new_velocity_limit_builder = cala_ledger::velocity::NewVelocityLimit::builder();
608 new_velocity_limit_builder
609 .id(input.velocity_limit_id)
610 .name(input.name)
611 .description(input.description);
612
613 if let Some(condition) = input.condition {
614 new_velocity_limit_builder.condition(condition);
615 }
616
617 if let Some(currency) = input.currency {
618 new_velocity_limit_builder.currency(currency);
619 }
620
621 let mut new_window = Vec::new();
622 for partition_key_input in input.window {
623 let mut new_partition_key_builder = cala_ledger::velocity::NewPartitionKey::builder();
624 let partition_key = new_partition_key_builder
625 .alias(partition_key_input.alias)
626 .value(partition_key_input.value)
627 .build()?;
628
629 new_window.push(partition_key);
630 }
631 new_velocity_limit_builder.window(new_window);
632
633 let mut new_limit_builder = cala_ledger::velocity::NewLimit::builder();
634
635 if let Some(timestamp_source) = input.limit.timestamp_source {
636 new_limit_builder.timestamp_source(timestamp_source);
637 }
638
639 let mut new_balance_limits = Vec::new();
640 for balance_limit_input in input.limit.balance {
641 let mut new_balance_limit_builder = cala_ledger::velocity::NewBalanceLimit::builder();
642 new_balance_limit_builder
643 .limit_type(balance_limit_input.limit_type)
644 .layer(balance_limit_input.layer)
645 .amount(balance_limit_input.amount)
646 .enforcement_direction(balance_limit_input.normal_balance_type);
647
648 if let Some(start) = balance_limit_input.start {
649 new_balance_limit_builder.start(start);
650 }
651
652 if let Some(end) = balance_limit_input.end {
653 new_balance_limit_builder.end(end);
654 }
655
656 let new_balance_limit = new_balance_limit_builder.build()?;
657 new_balance_limits.push(new_balance_limit);
658 }
659 new_limit_builder.balance(new_balance_limits);
660 let new_limit = new_limit_builder.build()?;
661
662 new_velocity_limit_builder.limit(new_limit);
663
664 if let Some(params) = input.params {
665 let mut new_params = Vec::new();
666 for param in params {
667 let mut param_builder = NewParamDefinition::builder();
668 param_builder.name(param.name).r#type(param.r#type.into());
669 if let Some(default) = param.default {
670 param_builder.default_expr(default);
671 }
672 if let Some(description) = param.description {
673 param_builder.description(description);
674 }
675 let new_param = param_builder.build()?;
676 new_params.push(new_param);
677 }
678 new_velocity_limit_builder.params(new_params);
679 }
680
681 let new_velocity_limit = new_velocity_limit_builder.build()?;
682
683 let velocity_limit = app
684 .ledger()
685 .velocities()
686 .create_limit_in_op(&mut op, new_velocity_limit)
687 .await?;
688
689 Ok(velocity_limit.into())
690 }
691
692 async fn velocity_control_create(
693 &self,
694 ctx: &Context<'_>,
695 input: VelocityControlCreateInput,
696 ) -> Result<VelocityControlCreatePayload> {
697 let app = ctx.data_unchecked::<CalaApp>();
698 let mut op = ctx
699 .data_unchecked::<DbOp>()
700 .try_lock()
701 .expect("Lock held concurrently");
702
703 let mut new_velocity_control_builder = cala_ledger::velocity::NewVelocityControl::builder();
704 new_velocity_control_builder
705 .id(input.velocity_control_id)
706 .name(input.name)
707 .description(input.description);
708
709 if let Some(condition) = input.condition {
710 new_velocity_control_builder.condition(condition);
711 }
712
713 let mut new_velocity_enforcement_builder =
714 cala_ledger::velocity::NewVelocityEnforcement::builder();
715 new_velocity_enforcement_builder.action(input.enforcement.velocity_enforcement_action);
716 let new_velocity_enforcement = new_velocity_enforcement_builder.build()?;
717
718 new_velocity_control_builder.enforcement(new_velocity_enforcement);
719
720 let new_velocity_control = new_velocity_control_builder.build()?;
721 let velocity_control = app
722 .ledger()
723 .velocities()
724 .create_control_in_op(&mut op, new_velocity_control)
725 .await?;
726
727 Ok(velocity_control.into())
728 }
729
730 async fn velocity_control_add_limit(
731 &self,
732 ctx: &Context<'_>,
733 input: VelocityControlAddLimitInput,
734 ) -> Result<VelocityControlAddLimitPayload> {
735 let app = ctx.data_unchecked::<CalaApp>();
736 let mut op = ctx
737 .data_unchecked::<DbOp>()
738 .try_lock()
739 .expect("Lock held concurrently");
740
741 let velocity_limit = app
742 .ledger()
743 .velocities()
744 .add_limit_to_control_in_op(
745 &mut op,
746 input.velocity_control_id.into(),
747 input.velocity_limit_id.into(),
748 )
749 .await?;
750
751 Ok(velocity_limit.into())
752 }
753
754 async fn velocity_control_attach(
755 &self,
756 ctx: &Context<'_>,
757 input: VelocityControlAttachInput,
758 ) -> Result<VelocityControlAttachPayload> {
759 let app = ctx.data_unchecked::<CalaApp>();
760 let mut op = ctx
761 .data_unchecked::<DbOp>()
762 .try_lock()
763 .expect("Lock held concurrently");
764 let params = cala_ledger::tx_template::Params::from(input.params);
765
766 let velocity_control = app
767 .ledger()
768 .velocities()
769 .attach_control_to_account_in_op(
770 &mut op,
771 input.velocity_control_id.into(),
772 input.account_id.into(),
773 params,
774 )
775 .await?;
776
777 Ok(velocity_control.into())
778 }
779}