feat: add transaction

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

@@ -7,4 +7,5 @@ pub mod account;
 pub mod journal;
 pub mod outbox;
 pub mod primitives;
+pub mod transaction;
 pub mod tx_template;

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

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-use crate::{account::*, journal::*, primitives::*, tx_template::*};
+use crate::{account::*, journal::*, primitives::*, transaction::*, tx_template::*};
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct OutboxEvent {
@@ -37,6 +37,10 @@ pub enum OutboxEventPayload {
         source: DataSource,
         tx_template: TxTemplateValues,
     },
+    TransactionCreated {
+        source: DataSource,
+        transaction: TransactionValues,
+    },
 }
 
 #[derive(

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

@@ -6,6 +6,7 @@ crate::entity_id! { AccountId }
 crate::entity_id! { JournalId }
 crate::entity_id! { DataSourceId }
 crate::entity_id! { TxTemplateId }
+crate::entity_id! { TransactionId }
 
 #[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, sqlx::Type)]
 #[sqlx(type_name = "DebitOrCredit", rename_all = "snake_case")]

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

@@ -0,0 +1,15 @@
+use serde::{Deserialize, Serialize};
+
+use super::primitives::*;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct TransactionValues {
+    pub id: TransactionId,
+    pub journal_id: JournalId,
+    pub tx_template_id: TxTemplateId,
+    pub effective: chrono::NaiveDate,
+    pub correlation_id: String,
+    pub external_id: Option<String>,
+    pub description: Option<String>,
+    pub metadata: Option<serde_json::Value>,
+}

cala-ledger-outbox-client/Cargo.toml

@@ -20,6 +20,7 @@ prost-wkt-types = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
+chrono = { workspace = true }
 
 [build-dependencies]
 protobuf-src = { version = "1.1.0" }

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

@@ -1,4 +1,6 @@
-use cala_types::{account::*, journal::*, outbox::*, primitives::*, tx_template::*};
+use cala_types::{
+    account::*, journal::*, outbox::*, primitives::*, transaction::*, tx_template::*,
+};
 use cel_interpreter::CelExpression;
 
 use crate::{client::proto, error::*};
@@ -57,6 +59,15 @@ impl TryFrom<proto::cala_ledger_event::Payload> for OutboxEventPayload {
                     tx_template.ok_or(CalaLedgerOutboxClientError::MissingField)?,
                 )?,
             },
+            proto::cala_ledger_event::Payload::TransactionCreated(proto::TransactionCreated {
+                data_source_id,
+                transaction,
+            }) => TransactionCreated {
+                source: data_source_id.parse()?,
+                transaction: TransactionValues::try_from(
+                    transaction.ok_or(CalaLedgerOutboxClientError::MissingField)?,
+                )?,
+            },
             proto::cala_ledger_event::Payload::Empty(_) => Empty,
         };
         Ok(res)
@@ -246,3 +257,31 @@ impl From<proto::ParamDataType> for ParamDataType {
         }
     }
 }
+
+impl TryFrom<proto::Transaction> for TransactionValues {
+    type Error = CalaLedgerOutboxClientError;
+    fn try_from(
+        proto::Transaction {
+            id,
+            journal_id,
+            tx_template_id,
+            effective,
+            correlation_id,
+            external_id,
+            description,
+            metadata,
+        }: proto::Transaction,
+    ) -> Result<Self, Self::Error> {
+        let res = Self {
+            id: id.parse()?,
+            journal_id: journal_id.parse()?,
+            tx_template_id: tx_template_id.parse()?,
+            effective: effective.parse()?,
+            correlation_id,
+            external_id,
+            description: description.map(String::from),
+            metadata: metadata.map(serde_json::to_value).transpose()?,
+        };
+        Ok(res)
+    }
+}

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

@@ -19,4 +19,6 @@ pub enum CalaLedgerOutboxClientError {
     ParseTagError(#[from] cala_types::primitives::ParseTagError),
     #[error("CalaLedgerOutboxError - CelError: {0}")]
     CelError(#[from] cel_interpreter::CelError),
+    #[error("CalaLedgerOutboxError - ChronoError: {0}")]
+    ChronoError(#[from] chrono::ParseError),
 }

cala-ledger/migrations/20231208110808_cala_ledger_setup.sql

@@ -71,6 +71,28 @@ CREATE TABLE cala_tx_template_events (
   FOREIGN KEY (data_source_id, id) REFERENCES cala_tx_templates(data_source_id, id)
 );
 
+CREATE TABLE cala_transactions (
+  data_source_id UUID NOT NULL DEFAULT '00000000-0000-0000-0000-000000000000',
+  id UUID NOT NULL,
+  journal_id UUID NOT NULL,
+  external_id VARCHAR DEFAULT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  UNIQUE(data_source_id, id),
+  FOREIGN KEY (data_source_id, journal_id) REFERENCES cala_journals(data_source_id, id)
+);
+CREATE UNIQUE INDEX idx_cala_transactions_data_source_id_external_id ON cala_transactions (data_source_id, external_id) WHERE external_id IS NOT NULL;
+
+CREATE TABLE cala_transaction_events (
+  data_source_id UUID NOT NULL DEFAULT '00000000-0000-0000-0000-000000000000',
+  id UUID NOT NULL,
+  sequence INT NOT NULL,
+  event_type VARCHAR NOT NULL,
+  event JSONB NOT NULL,
+  recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  UNIQUE(data_source_id, id, sequence),
+  FOREIGN KEY (data_source_id, id) REFERENCES cala_transactions(data_source_id, id)
+);
+
 CREATE TABLE cala_outbox_events (
   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
   sequence BIGSERIAL UNIQUE,

cala-ledger/src/ledger/error.rs

@@ -4,6 +4,7 @@ use crate::{
     account::error::AccountError,
     journal::error::JournalError,
     outbox::{error::OutboxError, server::error::OutboxServerError},
+    transaction::error::TransactionError,
     tx_template::error::TxTemplateError,
 };
 
@@ -25,4 +26,6 @@ pub enum LedgerError {
     JournalError(#[from] JournalError),
     #[error("LedgerError - TxTemplateError: {0}")]
     TxTemplateError(#[from] TxTemplateError),
+    #[error("LedgerError - TransactionError: {0}")]
+    TransactionError(#[from] TransactionError),
 }

cala-ledger/src/ledger/mod.rs

@@ -11,6 +11,7 @@ use crate::{
     account::Accounts,
     journal::Journals,
     outbox::{server, EventSequence, Outbox, OutboxListener},
+    transaction::Transactions,
     tx_template::TxTemplates,
 };
 #[cfg(feature = "import")]
@@ -27,6 +28,7 @@ pub struct CalaLedger {
     _pool: PgPool,
     accounts: Accounts,
     journals: Journals,
+    transactions: Transactions,
     tx_templates: TxTemplates,
     outbox: Outbox,
     #[allow(clippy::type_complexity)]
@@ -63,11 +65,13 @@ impl CalaLedger {
         let accounts = Accounts::new(&pool, outbox.clone());
         let journals = Journals::new(&pool, outbox.clone());
         let tx_templates = TxTemplates::new(&pool, outbox.clone());
+        let transactions = Transactions::new(&pool, outbox.clone());
         Ok(Self {
             accounts,
             journals,
             tx_templates,
             outbox,
+            transactions,
             outbox_handle: Arc::new(Mutex::new(outbox_handle)),
             _pool: pool,
         })
@@ -85,6 +89,10 @@ impl CalaLedger {
         &self.tx_templates
     }
 
+    pub fn transactions(&self) -> &Transactions {
+        &self.transactions
+    }
+
     pub async fn register_outbox_listener(
         &self,
         start_after: Option<EventSequence>,
@@ -114,6 +122,11 @@ impl CalaLedger {
                     .sync_journal_creation(tx, origin, journal)
                     .await?
             }
+            TransactionCreated { transaction, .. } => {
+                self.transactions
+                    .sync_transaction_creation(tx, origin, transaction)
+                    .await?
+            }
             TxTemplateCreated { tx_template, .. } => {
                 self.tx_templates
                     .sync_tx_template_creation(tx, origin, tx_template)

cala-ledger/src/lib.rs

@@ -5,6 +5,7 @@ pub mod account;
 pub mod entity;
 pub mod journal;
 pub mod migrate;
+pub mod transaction;
 pub mod tx_template;
 
 mod ledger;

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

@@ -7,6 +7,7 @@ use crate::{
         error::OutboxError,
         event::{OutboxEvent, OutboxEventPayload},
     },
+    transaction::TransactionValues,
     tx_template::*,
 };
 
@@ -41,6 +42,13 @@ impl From<OutboxEvent> for proto::CalaLedgerEvent {
                 data_source_id: source.to_string(),
                 tx_template: Some(proto::TxTemplate::from(tx_template)),
             }),
+            OutboxEventPayload::TransactionCreated {
+                source,
+                transaction,
+            } => proto::cala_ledger_event::Payload::TransactionCreated(proto::TransactionCreated {
+                data_source_id: source.to_string(),
+                transaction: Some(proto::Transaction::from(transaction)),
+            }),
             OutboxEventPayload::Empty => proto::cala_ledger_event::Payload::Empty(true),
         };
         proto::CalaLedgerEvent {
@@ -236,6 +244,34 @@ impl From<TxInput> for proto::TxInput {
     }
 }
 
+impl From<TransactionValues> for proto::Transaction {
+    fn from(
+        TransactionValues {
+            id,
+            journal_id,
+            tx_template_id,
+            correlation_id,
+            external_id,
+            effective,
+            description,
+            metadata,
+        }: TransactionValues,
+    ) -> Self {
+        proto::Transaction {
+            id: id.to_string(),
+            journal_id: journal_id.to_string(),
+            tx_template_id: tx_template_id.to_string(),
+            correlation_id,
+            external_id,
+            effective: effective.to_string(),
+            description: description.map(String::from),
+            metadata: metadata.map(|json| {
+                serde_json::from_value(json).expect("Could not transfer json -> struct")
+            }),
+        }
+    }
+}
+
 impl From<OutboxError> for tonic::Status {
     fn from(err: OutboxError) -> Self {
         // match err {