feat: integrate session persistence into payjoin workflow

- Add database initialization in handlers
- Replace NoopSessionPersister with real persisters
- Implement session resumption for existing sessions
- Add input-seen-before tracking in receiver flow
- Remove verbose error wrapping (use ? operator)

Author: Mshehu5

src/handlers.rs

@@ -654,6 +654,8 @@ pub(crate) async fn handle_online_wallet_subcommand(
     wallet: &mut Wallet,
     client: &BlockchainClient,
     online_subcommand: OnlineWalletSubCommand,
+    datadir: Option<std::path::PathBuf>,
+    wallet_name: String,
 ) -> Result<String, Error> {
     match online_subcommand {
         FullScan {
@@ -771,7 +773,8 @@ pub(crate) async fn handle_online_wallet_subcommand(
             max_fee_rate,
         } => {
             let relay_manager = Arc::new(Mutex::new(RelayManager::new()));
-            let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager);
+            let db = open_payjoin_db(datadir.clone(), &wallet_name)?;
+            let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager, db);
             return payjoin_manager
                 .receive_payjoin(amount, directory, max_fee_rate, ohttp_relay, client)
                 .await;
@@ -782,11 +785,25 @@ pub(crate) async fn handle_online_wallet_subcommand(
             fee_rate,
         } => {
             let relay_manager = Arc::new(Mutex::new(RelayManager::new()));
-            let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager);
+            let db = open_payjoin_db(datadir.clone(), &wallet_name)?;
+            let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager, db);
             return payjoin_manager
                 .send_payjoin(uri, fee_rate, ohttp_relay, client)
                 .await;
         }
+        ResumePayjoin {
+            directory,
+            ohttp_relay,
+            session_id,
+        } => {
+            let relay_manager = Arc::new(Mutex::new(RelayManager::new()));
+            let db = open_payjoin_db(datadir, &wallet_name)?;
+            let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager, db);
+            return payjoin_manager
+                .resume_payjoins(directory, ohttp_relay, session_id, client)
+                .await;
+        }
+        PayjoinHistory => crate::payjoin::PayjoinManager::history(datadir, &wallet_name),
     }
 }
 
@@ -1257,11 +1274,12 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
             wallet,
             subcommand: WalletSubCommand::OnlineWalletSubCommand(online_subcommand),
         } => {
-            let home_dir = prepare_home_dir(cli_opts.datadir)?;
+            let home_dir = prepare_home_dir(cli_opts.datadir.clone())?;
 
             let (wallet_opts, network) = load_wallet_config(&home_dir, &wallet)?;
 
             let database_path = prepare_wallet_db_dir(&home_dir, &wallet)?;
+            let wallet_name = wallet.clone();
 
             #[cfg(any(feature = "sqlite", feature = "redb"))]
             let result = {
@@ -1296,6 +1314,8 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
                     &mut wallet,
                     &blockchain_client,
                     online_subcommand,
+                    cli_opts.datadir.clone(),
+                    wallet_name.clone(),
                 )
                 .await?;
                 wallet.persist(&mut persister)?;
@@ -1306,8 +1326,14 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
                 let mut wallet = new_wallet(network, wallet_opts)?;
                 let blockchain_client =
                     crate::utils::new_blockchain_client(wallet_opts, &wallet, database_path)?;
-                handle_online_wallet_subcommand(&mut wallet, &blockchain_client, online_subcommand)
-                    .await?
+                handle_online_wallet_subcommand(
+                    &mut wallet,
+                    &blockchain_client,
+                    online_subcommand,
+                    cli_opts.datadir.clone(),
+                    wallet_name,
+                )
+                .await?
             };
             Ok(result)
         }
@@ -1510,9 +1536,15 @@ async fn respond(
         } => {
             let blockchain =
                 new_blockchain_client(wallet_opts, wallet, _datadir).map_err(|e| e.to_string())?;
-            let value = handle_online_wallet_subcommand(wallet, &blockchain, online_subcommand)
-                .await
-                .map_err(|e| e.to_string())?;
+            let value = handle_online_wallet_subcommand(
+                wallet,
+                &blockchain,
+                online_subcommand,
+                cli_opts.datadir.clone(),
+                wallet_name.clone(),
+            )
+            .await
+            .map_err(|e| e.to_string())?;
             Some(value)
         }
         ReplSubCommand::Wallet {