feat: add shutdown method to nodejs api

cala-ledger/src/ledger/mod.rs

@@ -3,6 +3,7 @@ pub mod error;
 
 use sqlx::PgPool;
 use std::sync::{Arc, Mutex};
+use tokio::{select, sync::oneshot};
 
 pub use config::*;
 use error::*;
@@ -19,6 +20,8 @@ pub struct CalaLedger {
     accounts: Accounts,
     journals: Journals,
     outbox_handle: Arc<Mutex<Option<tokio::task::JoinHandle<Result<(), LedgerError>>>>>,
+    abort_sender: Arc<Mutex<Option<oneshot::Sender<()>>>>,
+    abort_receiver: Arc<Mutex<Option<oneshot::Receiver<()>>>>,
 }
 
 impl CalaLedger {
@@ -48,12 +51,16 @@ impl CalaLedger {
             outbox_handle = Some(Self::start_outbox_server(outbox_config, outbox.clone()));
         }
 
+        let (abort_sender, abort_receiver) = oneshot::channel::<()>();
+
         let accounts = Accounts::new(&pool, outbox.clone());
         let journals = Journals::new(&pool, outbox);
         Ok(Self {
             accounts,
             journals,
             outbox_handle: Arc::new(Mutex::new(outbox_handle)),
+            abort_sender: Arc::new(Mutex::new(Some(abort_sender))),
+            abort_receiver: Arc::new(Mutex::new(Some(abort_receiver))),
             _pool: pool,
         })
     }
@@ -67,17 +74,36 @@ impl CalaLedger {
     }
 
     pub async fn await_outbox_handle(&self) -> Result<(), LedgerError> {
-        let handle = { self.outbox_handle.lock().expect("poisened mutex").take() };
-        if let Some(handle) = handle {
-            return handle.await.expect("Couldn't await outbox handle");
+        let mut handle = match self.outbox_handle.lock().expect("poisened mutex").take() {
+            Some(handle) => handle,
+            None => return Ok(()),
+        };
+
+
+        let abort_receiver = match self.abort_receiver.lock().expect("poisened mutex").take() {
+            Some(abort_receiver) => abort_receiver,
+            None => return Ok(()),
+        };
+
+        select! {
+            result = (&mut handle) => {
+                result.expect("Couldn't await outbox handle")
+            },
+
+            _ = abort_receiver => {
+                handle.abort();
+                Ok(())
+            },
         }
-        Ok(())
     }
 
-    pub fn shutdown_outbox(&mut self) -> Result<(), LedgerError> {
-        if let Some(handle) = self.outbox_handle.lock().expect("poisened mutex").take() {
-            handle.abort();
-        }
+    pub fn shutdown_outbox(&self) -> Result<(), LedgerError> {
+        let abort_sender = match self.abort_sender.lock().expect("poisened mutex").take() {
+            Some(abort_sender) => abort_sender,
+            None => return Ok(()),
+        };
+
+        let _ = abort_sender.send(());
         Ok(())
     }
 

cala-nodejs/index.d.ts

@@ -3,37 +3,37 @@
 
 /* auto-generated by NAPI-RS */
 
-export interface NewAccount {
-  id?: string
+export interface AccountValues {
+  id: string
   code: string
   name: string
+  tags: Array<string>
   externalId?: string
   description?: string
-  tags?: Array<string>
   metadata?: any
 }
-export interface AccountValues {
-  id: string
+export interface NewAccount {
+  id?: string
   code: string
   name: string
-  tags: Array<string>
   externalId?: string
   description?: string
+  tags?: Array<string>
   metadata?: any
 }
 export interface PaginatedAccounts {
   accounts: Array<AccountValues>
   hasNextPage: boolean
   endCursor?: CursorToken
 }
-export interface NewJournal {
-  id?: string
+export interface JournalValues {
+  id: string
   name: string
   externalId?: string
   description?: string
 }
-export interface JournalValues {
-  id: string
+export interface NewJournal {
+  id?: string
   name: string
   externalId?: string
   description?: string
@@ -60,16 +60,19 @@ export class CalaAccounts {
 }
 export class CalaAccount {
   id(): string
+  values(): AccountValues
 }
 export class CalaJournals {
   create(newJournal: NewJournal): Promise<CalaJournal>
 }
 export class CalaJournal {
   id(): string
+  values(): JournalValues
 }
 export class CalaLedger {
   static connect(config: CalaLedgerConfig): Promise<CalaLedger>
   accounts(): CalaAccounts
   journals(): CalaJournals
+  shutdownOutbox(): Promise<void>
   awaitOutboxServer(): Promise<void>
 }

cala-nodejs/src/ledger/mod.rs

@@ -45,6 +45,14 @@ impl CalaLedger {
     Ok(CalaJournals::new(self.inner.journals()))
   }
 
+  #[napi]
+  pub async fn shutdown_outbox(&self) -> napi::Result<()> {
+    self
+      .inner
+      .shutdown_outbox()
+      .map_err(crate::generic_napi_error)
+  }
+
   #[napi]
   pub async fn await_outbox_server(&self) -> napi::Result<()> {
     self

examples/nodejs/src/index.ts

@@ -39,6 +39,23 @@ const main = async () => {
   })
 
   console.log("Journal Created", journal.id());
+
+  const shutdown = async () => {
+    console.log("Attempting shut down...")
+
+    let shutdown
+    try {
+      shutdown = await cala.shutdownOutbox()
+    } catch(err) {
+      console.log("Error when attempting to shutdown: ", { err })
+    }
+
+    console.log("Shutdown signal sent")
+  }
+
+  setTimeout(shutdown, 2000)
+  console.log("Starting outbox server...")
+  await cala.awaitOutboxServer()
 }
 
 main()