1use std::io::Write;
11use std::path::{Path, PathBuf};
12use std::str::FromStr;
13use std::sync::Arc;
14use std::time::{Duration, SystemTime, UNIX_EPOCH};
15
16use alloy::primitives::{Address, B256, TxHash, U256};
17use alloy::signers::local::PrivateKeySigner;
18use anyhow::Context as _;
19use decdn_client::buyer_pool::{
20 ToppedUpPool, ensure_allowance, escrowed_but_untracked, grade_deposit_credit, open_pool,
21 top_up, topped_up_effect,
22};
23use decdn_common::admin::{AdminRpcClient as _, BuyerPoolsResponse};
24use decdn_common::cli::{self, common::expand_tilde};
25use decdn_common::config::{
26 DEFAULT_CHAIN_ID, DEFAULT_REDEEM_INTERVAL_SECS, FileConfig, capability_expiry_margin_secs,
27 load_file_config,
28};
29use decdn_common::redact::{sanitize_err_chain, sanitize_error_sources};
30use decdn_incentive::buyer_pool::{BuyerLoad, BuyerPoolState, BuyerPoolStore};
31use decdn_incentive::buyer_pool_redb::{ReadOnlyBuyerPoolStore, RedbBuyerPoolStore};
32use decdn_incentive::eth_identity::{self, PasswordUse, load_signer};
33use decdn_incentive::payment_pool::{PaymentPool, enumerate_owned_pools};
34use decdn_incentive::{Capability, CapabilityGrant, Deployment, PoolId, voucher_domain};
35use serde::Serialize;
36
37use decdn_client::provider;
38
39use super::buyer_store::{
40 BuyerStoreOwner, classify_buyer_store, client_buyer_db, node_buyer_db,
41 open_client_store_for_escrow,
42};
43use super::chain_ctx;
44
45pub async fn pool_dispatch(args: &cli::PoolArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
47 match &args.command {
48 cli::PoolCommand::List(a) => list(a, config_path).await,
49 cli::PoolCommand::Open(a) => open(a, config_path).await,
50 cli::PoolCommand::TopUp(a) => top_up_cmd(a, config_path).await,
51 cli::PoolCommand::Close(a) => close(a, config_path).await,
52 cli::PoolCommand::Reclaim(a) => reclaim(a, config_path).await,
53 cli::PoolCommand::Assign(a) => assign(a, config_path).await,
54 }
55}
56
57#[derive(Debug)]
60struct Resolved {
61 rpc_url: String,
62 payment_pool: Address,
63 chain_id: u64,
64 data_dir: PathBuf,
65 keystore: PathBuf,
66 keystore_password_file: Option<PathBuf>,
70}
71
72fn resolve_data_dir(data_dir: Option<PathBuf>, file: &FileConfig) -> anyhow::Result<PathBuf> {
76 data_dir
77 .or_else(|| file.identity.as_ref().and_then(|i| i.data_dir.clone()))
78 .map(|p| expand_tilde(&p))
79 .or_else(cli::default_client_data_dir)
80 .ok_or_else(|| {
81 anyhow::anyhow!("data_dir not set and no default available (pass --data-dir)")
82 })
83}
84
85fn resolve_chain(args: &cli::PoolChainArgs, file: &FileConfig) -> anyhow::Result<Resolved> {
86 let bc = file.blockchain.as_ref();
87 let rpc_url = args
88 .rpc_url
89 .clone()
90 .or_else(|| bc.and_then(|b| b.rpc_url.clone()))
91 .ok_or_else(|| anyhow::anyhow!("rpc_url not set (--rpc-url or blockchain.rpc_url)"))?;
92 let payment_pool_raw = args
93 .payment_pool_address
94 .clone()
95 .or_else(|| bc.and_then(|b| b.payment_pool_address.clone()))
96 .ok_or_else(|| {
97 anyhow::anyhow!(
98 "payment_pool_address not set (--payment-pool-address or \
99 blockchain.payment_pool_address)"
100 )
101 })?;
102 let payment_pool =
103 super::chain_ctx::parse_nonzero_address(&payment_pool_raw, "payment_pool_address")?;
104 let chain_id = args
105 .chain_id
106 .or_else(|| bc.and_then(|b| b.chain_id))
107 .unwrap_or(DEFAULT_CHAIN_ID);
108 let data_dir = resolve_data_dir(args.data_dir.clone(), file)?;
109 let keystore = args
110 .keystore
111 .clone()
112 .or_else(|| bc.and_then(|b| b.eth_keystore.clone()))
113 .map_or_else(
114 || eth_identity::keystore_path(&data_dir),
115 |p| expand_tilde(&p),
116 );
117 Ok(Resolved {
118 rpc_url,
119 payment_pool,
120 chain_id,
121 data_dir,
122 keystore,
123 keystore_password_file: args.keystore_password_file.as_deref().map(expand_tilde),
124 })
125}
126
127fn load_buyer_signer(chain: &Resolved) -> anyhow::Result<PrivateKeySigner> {
131 let password = super::chain_ctx::read_keystore_password(
132 &super::chain_ctx::password_sources(
133 chain.keystore_password_file.as_deref(),
134 PasswordUse::Unlock,
135 ),
136 "eth keystore password",
137 )?
138 .into_secret();
139 load_signer(&chain.keystore, &password)
140}
141
142fn parse_pool_id(s: &str) -> anyhow::Result<PoolId> {
145 B256::from_str(s)
146 .map_err(|e| anyhow::anyhow!("invalid --pool {s:?}: expected a 32-byte hex hash: {e}"))
147}
148
149fn ensure_owned(on_chain_owner: Address, ours: Address, pool_id: PoolId) -> anyhow::Result<()> {
153 anyhow::ensure!(
154 on_chain_owner == ours,
155 "pool {pool_id} on-chain owner {on_chain_owner} is not this keystore's address {ours} \
156 — wrong keystore/contract, or a mistyped --pool"
157 );
158 Ok(())
159}
160
161async fn open(args: &cli::PoolOpenArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
165 let file = load_file_config(config_path)?;
166 let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
170 let store = open_client_store_for_escrow(&data_dir, "open a pool")?;
171 let chain = resolve_chain(&args.chain, &file)?;
172 let signer = Arc::new(load_buyer_signer(&chain)?);
173 let owner = signer.address();
174 let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
175 let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
176 let deployment = Deployment {
177 chain_id: chain.chain_id,
178 payment_pool: chain.payment_pool,
179 };
180
181 let token = contract
182 .usdc()
183 .call()
184 .await
185 .map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
186 let deposit = U256::from(args.deposit_micro_usdc);
187
188 ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(deposit)).await?;
189
190 let opened = open_pool(
191 &contract,
192 Arc::clone(&signer),
193 deployment,
194 token,
195 owner,
196 deposit,
197 )
198 .await?;
199
200 store
203 .record(&opened.state)
204 .map_err(|e| escrowed_but_untracked("buyer pool opened", opened.tx, e))?;
205
206 let out = PoolOpenJson {
207 pool_id: format!("{:#x}", opened.state.pool_id),
208 };
209 serde_json::to_writer_pretty(std::io::stdout(), &out)?;
210 println!();
211 Ok(())
212}
213
214#[derive(Serialize)]
216struct PoolOpenJson {
217 #[serde(rename = "poolId")]
218 pool_id: String,
219}
220
221async fn top_up_cmd(args: &cli::PoolTopUpArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
223 let file = load_file_config(config_path)?;
224 let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
225 let store = open_client_store_for_escrow(&data_dir, "top up a pool")?;
226 let chain = resolve_chain(&args.chain, &file)?;
227 let pool_id = parse_pool_id(&args.pool)?;
228 let signer = Arc::new(load_buyer_signer(&chain)?);
229 let owner = signer.address();
230 let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
231 let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
232
233 let pool = contract
234 .getPool(pool_id)
235 .call()
236 .await
237 .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
238 ensure_owned(pool.owner, owner, pool_id)?;
239
240 let additional = U256::from(args.amount_micro_usdc);
241 let token = contract
244 .usdc()
245 .call()
246 .await
247 .map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
248 ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(additional)).await?;
249
250 let ToppedUpPool { credited, tx, .. } = top_up(&contract, owner, pool_id, additional).await?;
251
252 let effect = topped_up_effect(pool_id, credited);
259 let new_deposit =
260 grade_deposit_credit(store.add_deposit(owner, pool_id, credited), &effect, tx)?;
261 println!("topped up pool {pool_id} by {credited} µUSDC; deposit now {new_deposit} µUSDC");
262 Ok(())
263}
264
265#[derive(Debug, Clone, Copy)]
271enum LocalBookkeeping<'a> {
272 Store(&'a RedbBuyerPoolStore),
274 DaemonOwned(&'a Path),
279}
280
281impl LocalBookkeeping<'_> {
282 const fn new<'a>(
285 store: Option<&'a RedbBuyerPoolStore>,
286 buyer_db: &'a Path,
287 ) -> LocalBookkeeping<'a> {
288 match store {
289 Some(store) => LocalBookkeeping::Store(store),
290 None => LocalBookkeeping::DaemonOwned(buyer_db),
291 }
292 }
293
294 fn report_daemon_row_untouched(buyer_db: &Path, pool_id: PoolId, verb: &str) {
298 eprintln!(
299 "warning: pool {pool_id} {verb} on-chain, but the daemon's row in {} was NOT \
300 cleared — `decdn pool` writes only the client store, and a running decdn-node holds \
301 its own exclusively. Run `decdn node pools`: if that pool is listed, the daemon is \
302 still using it — restart decdn-node so its bootstrap reconciles against the chain.",
303 buyer_db.display()
304 );
305 }
306
307 fn forget_after_close(
313 self,
314 owner: Address,
315 pool_id: PoolId,
316 tx: TxHash,
317 reclaim_note: &str,
318 ) -> anyhow::Result<()> {
319 match self {
320 Self::Store(store) => grade_local_forget(
321 store.forget_if_pool(owner, pool_id),
322 pool_id,
323 tx,
324 reclaim_note,
325 ),
326 Self::DaemonOwned(buyer_db) => {
327 Self::report_daemon_row_untouched(buyer_db, pool_id, "closed");
328 Ok(())
329 }
330 }
331 }
332
333 fn forget_after_reclaim(self, owner: Address, pool_id: PoolId) {
336 match self {
337 Self::Store(store) => {
338 if let Err(e) = store.forget_if_pool(owner, pool_id) {
347 eprintln!(
348 "warning: pool {pool_id} reclaimed on-chain but clearing it from the \
349 local store failed: {e}; if the row survived, a later fetch reuses this \
350 closed pool and its vouchers are rejected — re-run `decdn pool reclaim \
351 --pool {pool_id}` to clear it"
352 );
353 }
354 }
355 Self::DaemonOwned(buyer_db) => {
356 Self::report_daemon_row_untouched(buyer_db, pool_id, "reclaimed");
357 }
358 }
359 }
360}
361
362fn grade_local_forget(
390 outcome: Result<bool, decdn_incentive::StoreError>,
391 pool_id: PoolId,
392 tx: TxHash,
393 reclaim_note: &str,
394) -> anyhow::Result<()> {
395 match outcome {
396 Ok(_) => Ok(()),
397 Err(e) => anyhow::bail!(
398 "pool {pool_id} closed on-chain (tx {tx}) but clearing it from the local store \
399 failed: {e}; the row may survive and send a later fetch back to this closing \
400 pool, whose vouchers stop being redeemable at the dispute deadline — \
401 {reclaim_note}, which also clears the row"
402 ),
403 }
404}
405
406#[derive(Debug, Clone, Copy, PartialEq, Eq)]
408enum TxOutcome {
409 Landed,
411 Reverted,
413}
414
415const fn receipt_outcome(landed: bool) -> TxOutcome {
418 if landed {
419 TxOutcome::Landed
420 } else {
421 TxOutcome::Reverted
422 }
423}
424
425#[derive(Debug, Clone, Copy, PartialEq, Eq)]
428enum ClosePlan {
429 Close,
431 Skip(&'static str),
433}
434
435#[derive(Debug, Clone, Copy, PartialEq, Eq)]
440enum ReclaimPlan {
441 Reclaim,
443 SkipOpen,
445 SkipInWindow(u64),
448 SkipClosed,
450 SkipUnknown,
453}
454
455#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
458struct BatchTally {
459 acted: usize,
461 skipped: usize,
463 failed: usize,
465}
466
467const fn plan_close(status: PaymentPool::Status) -> ClosePlan {
469 match status {
470 PaymentPool::Status::Open => ClosePlan::Close,
471 PaymentPool::Status::Closing => ClosePlan::Skip("already closing"),
472 PaymentPool::Status::Closed => ClosePlan::Skip("already closed"),
473 _ => ClosePlan::Skip("unknown on-chain status"),
476 }
477}
478
479const fn plan_reclaim(status: PaymentPool::Status, dispute_deadline: u64, now: u64) -> ReclaimPlan {
484 match status {
485 PaymentPool::Status::Open => ReclaimPlan::SkipOpen,
486 PaymentPool::Status::Closing if now >= dispute_deadline => ReclaimPlan::Reclaim,
487 PaymentPool::Status::Closing => ReclaimPlan::SkipInWindow(dispute_deadline),
488 PaymentPool::Status::Closed => ReclaimPlan::SkipClosed,
489 _ => ReclaimPlan::SkipUnknown,
492 }
493}
494
495fn single_reclaim_gate(
508 pool_id: PoolId,
509 owner: Address,
510 status: PaymentPool::Status,
511 dispute_deadline: u64,
512 now: u64,
513) -> anyhow::Result<()> {
514 anyhow::ensure!(
515 owner != Address::ZERO,
516 "pool {pool_id} does not exist on this PaymentPool contract — a mistyped --pool, or \
517 the wrong payment_pool address"
518 );
519 match plan_reclaim(status, dispute_deadline, now) {
520 ReclaimPlan::Reclaim => Ok(()),
521 ReclaimPlan::SkipOpen => anyhow::bail!(
522 "pool {pool_id} is still Open — run `decdn pool close --pool {pool_id}` first"
523 ),
524 ReclaimPlan::SkipInWindow(deadline) => anyhow::bail!(
525 "pool {pool_id} is still in its dispute window; reclaimable after {}",
526 format_expiry(deadline, now)
527 ),
528 ReclaimPlan::SkipClosed => {
529 anyhow::bail!("pool {pool_id} is already Closed — its residual was already reclaimed")
530 }
531 ReclaimPlan::SkipUnknown => {
532 anyhow::bail!("pool {pool_id} has an unrecognized on-chain status — nothing to reclaim")
533 }
534 }
535}
536
537fn batch_result(tally: &BatchTally, verb: &str) -> anyhow::Result<()> {
545 anyhow::ensure!(
546 tally.failed == 0,
547 "{} pool(s) failed to {verb}; see the summary above",
548 tally.failed
549 );
550 Ok(())
551}
552
553async fn close_and_forget<P>(
563 contract: &PaymentPool::PaymentPoolInstance<P>,
564 store: LocalBookkeeping<'_>,
565 owner: Address,
566 pool_id: PoolId,
567) -> anyhow::Result<String>
568where
569 P: alloy::providers::Provider + Clone,
570{
571 let pending = match contract.closePool(pool_id).send().await {
572 Ok(pending) => pending,
573 Err(e) if e.as_revert_data().is_some() => {
574 anyhow::bail!("closePool reverted on-chain for pool {pool_id}")
575 }
576 Err(e) => return Err(anyhow::Error::new(e).context("closePool send failed")),
577 };
578 let receipt = pending
579 .get_receipt()
580 .await
581 .map_err(|e| anyhow::Error::new(e).context("closePool receipt failed"))?;
582
583 match receipt_outcome(receipt.status()) {
584 TxOutcome::Landed => {
585 let deadline_note = match contract.getPool(pool_id).call().await {
592 Ok(p) => format!("after Unix {}", p.disputeDeadline),
593 Err(e) => format!(
594 "after the dispute window (couldn't read the exact deadline: {})",
595 sanitize_error_sources(&e)
596 ),
597 };
598 let reclaim_note = format!("run `decdn pool reclaim --pool {pool_id}` {deadline_note}");
601 store.forget_after_close(owner, pool_id, receipt.transaction_hash, &reclaim_note)?;
602 Ok(reclaim_note)
603 }
604 TxOutcome::Reverted => anyhow::bail!(
605 "closePool reverted on-chain for pool {pool_id} (it may have raced a concurrent \
606 close)"
607 ),
608 }
609}
610
611async fn reclaim_and_forget<P>(
623 contract: &PaymentPool::PaymentPoolInstance<P>,
624 store: LocalBookkeeping<'_>,
625 owner: Address,
626 pool_id: PoolId,
627) -> anyhow::Result<()>
628where
629 P: alloy::providers::Provider + Clone,
630{
631 let pending = match contract.reclaim(pool_id).send().await {
632 Ok(pending) => pending,
633 Err(e) if e.as_revert_data().is_some() => anyhow::bail!(reclaim_reverted(pool_id)),
634 Err(e) => return Err(anyhow::Error::new(e).context("reclaim send failed")),
635 };
636 let receipt = pending
637 .get_receipt()
638 .await
639 .map_err(|e| anyhow::Error::new(e).context("reclaim receipt failed"))?;
640
641 match receipt_outcome(receipt.status()) {
642 TxOutcome::Landed => {
643 store.forget_after_reclaim(owner, pool_id);
644 Ok(())
645 }
646 TxOutcome::Reverted => anyhow::bail!(reclaim_reverted(pool_id)),
647 }
648}
649
650fn reclaim_reverted(pool_id: PoolId) -> String {
653 format!(
654 "reclaim reverted on-chain for pool {pool_id} after the pre-check passed (most likely \
655 a concurrent reclaim; the USDC refund transfer to the owner can also revert)"
656 )
657}
658
659async fn close(args: &cli::PoolCloseArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
664 let file = load_file_config(config_path)?;
665 let target = args.pool.as_deref().map(parse_pool_id).transpose()?;
669
670 let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
671 if target.is_none() {
672 store_owner.refuse_sweep("close")?;
675 }
676 let chain = resolve_chain(&args.chain, &file)?;
677 let store = store_owner.open_for_write()?;
678 let buyer_db = node_buyer_db(&chain.data_dir);
679 let books = LocalBookkeeping::new(store.as_ref(), &buyer_db);
680 let signer = Arc::new(load_buyer_signer(&chain)?);
681 let owner = signer.address();
682 let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
683 let contract = PaymentPool::new(chain.payment_pool, rpc);
684
685 let Some(pool_id) = target else {
686 return close_all(&contract, books, owner).await;
687 };
688
689 let pool = contract
690 .getPool(pool_id)
691 .call()
692 .await
693 .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
694 ensure_owned(pool.owner, owner, pool_id)?;
695 anyhow::ensure!(
696 matches!(pool.status, PaymentPool::Status::Open),
697 "pool {pool_id} is not Open — nothing to close (already closing or closed)"
698 );
699
700 let reclaim_note = close_and_forget(&contract, books, owner, pool_id).await?;
701 println!("closed pool {pool_id}; dispute window open — {reclaim_note}");
702 Ok(())
703}
704
705async fn close_all<P>(
709 contract: &PaymentPool::PaymentPoolInstance<P>,
710 store: LocalBookkeeping<'_>,
711 owner: Address,
712) -> anyhow::Result<()>
713where
714 P: alloy::providers::Provider + Clone,
715{
716 let ids = enumerate_owned_pools(contract, owner).await?;
717 if ids.is_empty() {
718 println!("no pools found for this keystore — nothing to close");
719 return Ok(());
720 }
721
722 let mut tally = BatchTally::default();
723 for pool_id in ids {
724 let status = match contract.getPool(pool_id).call().await {
725 Ok(pool) => pool.status,
726 Err(e) => {
727 eprintln!(
728 "failed to read pool {pool_id}: {}",
729 sanitize_error_sources(&e)
730 );
731 tally.failed += 1;
732 continue;
733 }
734 };
735 match plan_close(status) {
736 ClosePlan::Close => match close_and_forget(contract, store, owner, pool_id).await {
737 Ok(note) => {
738 println!("closed pool {pool_id}; dispute window open — {note}");
739 tally.acted += 1;
740 }
741 Err(e) => {
742 eprintln!("failed to close pool {pool_id}: {}", sanitize_err_chain(&e));
743 tally.failed += 1;
744 }
745 },
746 ClosePlan::Skip(reason) => {
747 println!("skipped pool {pool_id}: {reason}");
748 tally.skipped += 1;
749 }
750 }
751 }
752 println!(
753 "close --all: closed {}, skipped {}, failed {}",
754 tally.acted, tally.skipped, tally.failed
755 );
756 batch_result(&tally, "close")
757}
758
759async fn reclaim(args: &cli::PoolReclaimArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
765 let file = load_file_config(config_path)?;
766 let target = args.pool.as_deref().map(parse_pool_id).transpose()?;
770
771 let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
772 if target.is_none() {
773 store_owner.refuse_sweep("reclaim")?;
774 }
775 let chain = resolve_chain(&args.chain, &file)?;
776 if let Some(pool_id) = target {
777 let reader = PaymentPool::new(
781 chain.payment_pool,
782 provider::build_read_provider(&chain.rpc_url)?,
783 );
784 let pool = reader
785 .getPool(pool_id)
786 .call()
787 .await
788 .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
789 let now = chain_ctx::head_timestamp(reader.provider()).await?;
790 single_reclaim_gate(pool_id, pool.owner, pool.status, pool.disputeDeadline, now)?;
791 }
792 let store = store_owner.open_for_write()?;
793 let buyer_db = node_buyer_db(&chain.data_dir);
794 let books = LocalBookkeeping::new(store.as_ref(), &buyer_db);
795 let signer = Arc::new(load_buyer_signer(&chain)?);
796 let owner = signer.address();
797 let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
798 let contract = PaymentPool::new(chain.payment_pool, rpc);
799
800 let Some(pool_id) = target else {
801 return reclaim_all(&contract, books, owner).await;
802 };
803
804 reclaim_and_forget(&contract, books, owner, pool_id).await?;
805 println!("reclaimed pool {pool_id}; residual deposit refunded to its owner");
806 Ok(())
807}
808
809async fn reclaim_all<P>(
815 contract: &PaymentPool::PaymentPoolInstance<P>,
816 store: LocalBookkeeping<'_>,
817 owner: Address,
818) -> anyhow::Result<()>
819where
820 P: alloy::providers::Provider + Clone,
821{
822 let ids = enumerate_owned_pools(contract, owner).await?;
823 if ids.is_empty() {
824 println!("no pools found for this keystore — nothing to reclaim");
825 return Ok(());
826 }
827 let now = chain_ctx::head_timestamp(contract.provider()).await?;
828
829 let mut tally = BatchTally::default();
830 for pool_id in ids {
831 let pool = match contract.getPool(pool_id).call().await {
832 Ok(pool) => pool,
833 Err(e) => {
834 eprintln!(
835 "failed to read pool {pool_id}: {}",
836 sanitize_error_sources(&e)
837 );
838 tally.failed += 1;
839 continue;
840 }
841 };
842 match plan_reclaim(pool.status, pool.disputeDeadline, now) {
843 ReclaimPlan::Reclaim => {
844 match reclaim_and_forget(contract, store, owner, pool_id).await {
845 Ok(()) => {
846 println!(
847 "reclaimed pool {pool_id}; residual deposit refunded to its owner"
848 );
849 tally.acted += 1;
850 }
851 Err(e) => {
852 eprintln!(
853 "failed to reclaim pool {pool_id}: {}",
854 sanitize_err_chain(&e)
855 );
856 tally.failed += 1;
857 }
858 }
859 }
860 ReclaimPlan::SkipOpen => {
861 println!("skipped pool {pool_id}: still Open — run `pool close` first");
862 tally.skipped += 1;
863 }
864 ReclaimPlan::SkipInWindow(deadline) => {
865 println!(
866 "skipped pool {pool_id}: dispute window open — reclaimable after {}",
867 format_expiry(deadline, now)
868 );
869 tally.skipped += 1;
870 }
871 ReclaimPlan::SkipClosed => {
872 println!("skipped pool {pool_id}: already reclaimed");
873 tally.skipped += 1;
874 }
875 ReclaimPlan::SkipUnknown => {
876 println!("skipped pool {pool_id}: unrecognized on-chain status");
877 tally.skipped += 1;
878 }
879 }
880 }
881 println!(
882 "reclaim --all: reclaimed {}, skipped {}, failed {}",
883 tally.acted, tally.skipped, tally.failed
884 );
885 batch_result(&tally, "reclaim")
886}
887
888fn resolve_expiry(
894 now: u64,
895 expiry_secs: Option<u64>,
896 expiry_at: Option<u64>,
897) -> anyhow::Result<u64> {
898 let expiry = match (expiry_secs, expiry_at) {
899 (Some(secs), None) => now.checked_add(secs).ok_or_else(|| {
900 anyhow::anyhow!("--expiry-secs {secs} overflows the Unix epoch from now ({now})")
901 })?,
902 (None, Some(at)) => at,
903 _ => anyhow::bail!("exactly one of --expiry-secs or --expiry-at is required"),
905 };
906 anyhow::ensure!(
907 expiry > now,
908 "capability expiry {expiry} is not in the future (now is {now}); it would be dead on \
909 arrival — pick a later --expiry-at or a positive --expiry-secs"
910 );
911 Ok(expiry)
912}
913
914const DEFAULT_NODE_EXPIRY_MARGIN_SECS: u64 =
916 capability_expiry_margin_secs(DEFAULT_REDEEM_INTERVAL_SECS);
917
918const fn expires_within_default_node_margin(expiry: u64, now: u64) -> bool {
923 expiry.saturating_sub(now) <= DEFAULT_NODE_EXPIRY_MARGIN_SECS
924}
925
926fn unix_now() -> anyhow::Result<u64> {
934 SystemTime::now()
935 .duration_since(UNIX_EPOCH)
936 .map(|d| d.as_secs())
937 .map_err(|e| anyhow::anyhow!("the system clock is before the Unix epoch: {e}"))
938}
939
940const HEAD_READ_TIMEOUT: Duration = Duration::from_secs(5);
943
944async fn issuance_now(rpc_url: &str, timeout: Duration) -> anyhow::Result<u64> {
958 let head = match provider::build_read_provider(rpc_url) {
959 Ok(reader) => tokio::time::timeout(timeout, chain_ctx::head_timestamp(&reader))
960 .await
961 .unwrap_or_else(|_| Err(anyhow::anyhow!("no answer within {timeout:?}"))),
962 Err(e) => Err(e),
963 };
964 match head {
965 Ok(now) => Ok(now),
966 Err(e) => {
967 eprintln!(
968 "warning: could not read the chain head time ({}); checking the capability \
969 expiry against the local clock instead",
970 sanitize_err_chain(&e)
971 );
972 unix_now()
973 }
974 }
975}
976
977fn format_expiry(expiry: u64, now: u64) -> String {
981 let remaining = expiry.saturating_sub(now);
982 let days = remaining / 86_400;
983 let hours = (remaining % 86_400) / 3_600;
984 let mins = (remaining % 3_600) / 60;
985 format!("Unix {expiry} (in {days}d {hours}h {mins}m)")
986}
987
988async fn assign(args: &cli::PoolAssignArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
1001 let file = load_file_config(config_path)?;
1002 let chain = resolve_chain(&args.chain, &file)?;
1003 let pool_id = parse_pool_id(&args.pool)?;
1004 let signer_addr = super::chain_ctx::parse_nonzero_address(&args.signer, "--signer")?;
1005
1006 let now = issuance_now(&chain.rpc_url, HEAD_READ_TIMEOUT).await?;
1007 let expiry = resolve_expiry(now, args.expiry_secs, args.expiry_at)?;
1008 if expires_within_default_node_margin(expiry, now) {
1009 eprintln!(
1010 "warning: the capability expires in {}s; a node with the default redeem interval \
1011 refuses vouchers within {DEFAULT_NODE_EXPIRY_MARGIN_SECS}s of a capability's \
1012 expiry, so such a node serves this delegate nothing",
1013 expiry.saturating_sub(now)
1014 );
1015 }
1016
1017 let owner_signer = load_buyer_signer(&chain)?;
1018 let owner = owner_signer.address();
1019 let domain = voucher_domain(chain.chain_id, chain.payment_pool);
1020 let spending_cap = args.cap_micro_usdc;
1021
1022 let capability = Capability {
1023 signer: signer_addr,
1024 spending_cap,
1025 pool_id,
1026 expiry,
1027 };
1028 let signed = capability.sign(&owner_signer, &domain)?;
1029 let grant = CapabilityGrant::from_signed_capability(&signed);
1030 let token = grant.to_token();
1031
1032 let recovered = grant
1036 .owner(&domain)
1037 .map_err(|e| anyhow::anyhow!("re-recovering the owner from the fresh token failed: {e}"))?;
1038
1039 match provider::build_provider(&chain.rpc_url, &owner_signer) {
1045 Ok(rpc) => {
1046 let contract = PaymentPool::new(chain.payment_pool, rpc);
1047 ensure_on_chain_owner(&contract, pool_id, owner).await?;
1048 }
1049 Err(e) => eprintln!(
1050 "warning: could not build an RPC provider to check the on-chain owner of pool \
1051 {pool_id} ({}); issuing anyway — the node verifies the owner signature at redemption",
1052 sanitize_err_chain(&e)
1053 ),
1054 }
1055
1056 println!("pool: {pool_id}");
1057 println!("delegate: {signer_addr} (the voucher-signing key this authorizes)");
1058 println!(
1059 "spending cap: {} USDC ({} µUSDC)",
1060 format_usdc_u256(U256::from(spending_cap)),
1061 args.cap_micro_usdc
1062 );
1063 println!("expiry: {}", format_expiry(expiry, now));
1064 println!("owner: {recovered} (recovered from the signature)");
1065 println!("Token (give this to the delegated client):");
1066 println!("{token}");
1067 Ok(())
1068}
1069
1070async fn ensure_on_chain_owner<P>(
1089 contract: &PaymentPool::PaymentPoolInstance<P>,
1090 pool_id: PoolId,
1091 expected: Address,
1092) -> anyhow::Result<()>
1093where
1094 P: alloy::providers::Provider + Clone,
1095{
1096 match contract.getPool(pool_id).call().await {
1097 Ok(pool) => grade_on_chain_owner(pool.owner, expected).into_result(pool_id, expected),
1098 Err(e) => {
1099 eprintln!(
1100 "warning: could not read pool {pool_id} on-chain to confirm ownership ({}); \
1101 issuing anyway — the node verifies the owner signature at redemption",
1102 sanitize_error_sources(&e)
1103 );
1104 Ok(())
1105 }
1106 }
1107}
1108
1109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1115enum OwnerVerdict {
1116 Owned,
1118 NoSuchPool,
1121 OwnedByOther(Address),
1123}
1124
1125fn grade_on_chain_owner(on_chain: Address, expected: Address) -> OwnerVerdict {
1128 if on_chain.is_zero() {
1129 OwnerVerdict::NoSuchPool
1130 } else if on_chain == expected {
1131 OwnerVerdict::Owned
1132 } else {
1133 OwnerVerdict::OwnedByOther(on_chain)
1134 }
1135}
1136
1137impl OwnerVerdict {
1138 fn into_result(self, pool_id: PoolId, expected: Address) -> anyhow::Result<()> {
1147 match self {
1148 Self::Owned => Ok(()),
1149 Self::NoSuchPool => anyhow::bail!(
1150 "pool {pool_id} does not exist on this PaymentPool contract; a capability for a \
1151 pool that was never opened is rejected at redemption, so no token was issued — \
1152 check --pool, --payment-pool-address, and --chain-id"
1153 ),
1154 Self::OwnedByOther(on_chain) => anyhow::bail!(
1155 "pool {pool_id} on-chain owner {on_chain} is not this keystore's address \
1156 {expected}; a capability signed by a non-owner is rejected at redemption, so no \
1157 token was issued — sign with the owner keystore, or check --pool"
1158 ),
1159 }
1160 }
1161}
1162
1163async fn list(args: &cli::PoolListArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
1174 let data_dir = match args.chain.data_dir.clone() {
1175 Some(dir) => expand_tilde(&dir),
1176 None => resolve_data_dir(None, &load_file_config(config_path)?)?,
1177 };
1178
1179 if args.all {
1180 return list_all(args, config_path, &data_dir).await;
1181 }
1182
1183 match classify_buyer_store(&data_dir)? {
1184 BuyerStoreOwner::Node { data_dir, .. } => {
1185 list_from_daemon(args, config_path, &data_dir).await
1186 }
1187 BuyerStoreOwner::Client { data_dir } => list_from_client_store(args, &data_dir),
1188 }
1189}
1190
1191async fn list_all(
1209 args: &cli::PoolListArgs,
1210 config_path: Option<&Path>,
1211 data_dir: &Path,
1212) -> anyhow::Result<()> {
1213 let file = load_file_config(config_path)?;
1214 let chain = resolve_chain(&args.chain, &file)?;
1215 let signer = Arc::new(load_buyer_signer(&chain)?);
1216 let owner = signer.address();
1217 let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
1218 let contract = PaymentPool::new(chain.payment_pool, rpc);
1219
1220 let ids = enumerate_owned_pools(&contract, owner).await?;
1221 let tracked = tracked_pool_ids(args, config_path, data_dir).await;
1222 let now = chain_ctx::head_timestamp(contract.provider()).await?;
1223
1224 let mut rows = Vec::with_capacity(ids.len());
1225 for pool_id in ids {
1226 let pool = contract.getPool(pool_id).call().await.map_err(|e| {
1227 anyhow::anyhow!("getPool({pool_id}) failed: {}", sanitize_error_sources(&e))
1228 })?;
1229 rows.push(ChainPoolRow {
1230 pool_id,
1231 status: pool.status,
1232 deposit: U256::from(pool.deposit),
1233 total_redeemed: U256::from(pool.totalRedeemed),
1234 dispute_deadline: pool.disputeDeadline,
1235 tracked: tracked.as_ref().and_then(|local| local.verdict(pool_id)),
1236 });
1237 }
1238
1239 if let Some(local) = tracked.as_ref() {
1243 write_skipped_pools(&mut std::io::stderr().lock(), &local.undecodable())?;
1244 }
1245
1246 let mut out = std::io::stdout().lock();
1247 if args.json {
1248 let view = PoolListAllJson {
1249 source: SOURCE_CHAIN,
1250 owner: format!("{owner:#x}"),
1251 chain_id: chain.chain_id,
1252 payment_pool: format!("{:#x}", chain.payment_pool),
1253 local_store_read: tracked.is_some(),
1254 pools: rows.iter().map(|r| ChainPoolJson::at(r, now)).collect(),
1255 };
1256 serde_json::to_writer_pretty(&mut out, &view)?;
1257 writeln!(out)?;
1258 } else {
1259 write_chain_pools(
1260 &mut out,
1261 owner,
1262 chain.chain_id,
1263 &rows,
1264 now,
1265 tracked.is_some(),
1266 )?;
1267 }
1268 Ok(())
1269}
1270
1271#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1273enum RowState {
1274 Decoded,
1276 Undecodable,
1280}
1281
1282#[derive(Debug, Default)]
1292struct TrackedPools(std::collections::BTreeMap<PoolId, RowState>);
1293
1294impl TrackedPools {
1295 fn verdict(&self, pool_id: PoolId) -> Option<bool> {
1299 match self.0.get(&pool_id) {
1300 Some(RowState::Decoded) => Some(true),
1301 Some(RowState::Undecodable) => None,
1302 None => Some(false),
1303 }
1304 }
1305
1306 fn undecodable(&self) -> Vec<PoolId> {
1308 self.0
1309 .iter()
1310 .filter(|(_, state)| **state == RowState::Undecodable)
1311 .map(|(id, _)| *id)
1312 .collect()
1313 }
1314
1315 fn from_wire(resp: &BuyerPoolsResponse) -> Option<Self> {
1324 let mut rows = std::collections::BTreeMap::new();
1325 for (id, state) in resp
1326 .pools
1327 .iter()
1328 .map(|p| (p.pool_id.as_str(), RowState::Decoded))
1329 .chain(
1330 resp.skipped
1331 .iter()
1332 .map(|p| (p.as_str(), RowState::Undecodable)),
1333 )
1334 {
1335 rows.insert(PoolId::from_str(id).ok()?, state);
1336 }
1337 Some(Self(rows))
1338 }
1339}
1340
1341impl From<BuyerLoad> for TrackedPools {
1342 fn from(load: BuyerLoad) -> Self {
1343 Self(
1344 load.pools
1345 .iter()
1346 .map(|p| (p.pool_id, RowState::Decoded))
1347 .chain(
1348 load.skipped
1349 .into_iter()
1350 .map(|id| (id, RowState::Undecodable)),
1351 )
1352 .collect(),
1353 )
1354 }
1355}
1356
1357async fn tracked_pool_ids(
1375 args: &cli::PoolListArgs,
1376 config_path: Option<&Path>,
1377 data_dir: &Path,
1378) -> Option<TrackedPools> {
1379 let owner = match classify_buyer_store(data_dir) {
1380 Ok(owner) => owner,
1381 Err(err) => {
1382 let mut w = std::io::stderr().lock();
1383 let _ = writeln!(
1384 w,
1385 "warning: {err:#}; every TRACKED value below is unknown, which is not the same \
1386 as untracked"
1387 );
1388 return None;
1389 }
1390 };
1391 match owner {
1392 BuyerStoreOwner::Client { data_dir } => read_local_pools(&client_buyer_db(&data_dir)),
1393 BuyerStoreOwner::Node { data_dir, .. } => {
1394 let buyer_db = node_buyer_db(&data_dir);
1395 match ReadOnlyBuyerPoolStore::open_file(&buyer_db) {
1396 Ok(reader) => load_local_pools(&reader, &buyer_db),
1397 Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => {
1402 daemon_pool_ids(args, config_path).await
1403 }
1404 Err(err) => {
1405 warn_unreadable_store(&buyer_db, &err);
1406 None
1407 }
1408 }
1409 }
1410 }
1411}
1412
1413fn read_local_pools(path: &Path) -> Option<TrackedPools> {
1419 match ReadOnlyBuyerPoolStore::open_file(path) {
1420 Ok(reader) => load_local_pools(&reader, path),
1421 Err(decdn_incentive::StoreError::Absent { .. }) => Some(TrackedPools::default()),
1422 Err(err) => {
1423 warn_unreadable_store(path, &err);
1424 None
1425 }
1426 }
1427}
1428
1429fn load_local_pools(reader: &ReadOnlyBuyerPoolStore, path: &Path) -> Option<TrackedPools> {
1432 match reader.load_all() {
1433 Ok(load) => Some(TrackedPools::from(load)),
1434 Err(err) => {
1435 warn_unreadable_store(path, &err);
1436 None
1437 }
1438 }
1439}
1440
1441fn warn_unreadable_store(path: &Path, err: &decdn_incentive::StoreError) {
1447 let mut w = std::io::stderr().lock();
1448 let _ = writeln!(
1449 w,
1450 "warning: the local buyer store at {} could not be read ({err}); every TRACKED value \
1451 below is unknown, which is not the same as untracked",
1452 path.display()
1453 );
1454}
1455
1456async fn daemon_pool_ids(
1463 args: &cli::PoolListArgs,
1464 config_path: Option<&Path>,
1465) -> Option<TrackedPools> {
1466 if args.timeout_ms == 0 {
1467 let mut w = std::io::stderr().lock();
1470 let _ = writeln!(
1471 w,
1472 "warning: --timeout-ms 0 skips the daemon lookup (jsonrpsee reads a zero duration \
1473 as 'never'), so every TRACKED value below is unknown"
1474 );
1475 return None;
1476 }
1477 let url =
1478 crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path).ok()?;
1479 let client = jsonrpsee::http_client::HttpClientBuilder::default()
1480 .request_timeout(std::time::Duration::from_millis(args.timeout_ms))
1481 .build(&url)
1482 .ok()?;
1483 let resp: BuyerPoolsResponse = client.pools().await.ok()?;
1484 let tracked = TrackedPools::from_wire(&resp);
1485 if tracked.is_none() {
1486 let mut w = std::io::stderr().lock();
1487 let _ = writeln!(
1488 w,
1489 "warning: the daemon reported a pool id this binary cannot parse, so every TRACKED \
1490 value below is unknown. The daemon and this CLI disagree on the wire format — \
1491 check that both are the same build."
1492 );
1493 }
1494 tracked
1495}
1496
1497fn list_from_client_store(args: &cli::PoolListArgs, data_dir: &Path) -> anyhow::Result<()> {
1499 let store_path = client_buyer_db(data_dir);
1500 let store = RedbBuyerPoolStore::open(data_dir)?;
1501 let load = store.load_all()?;
1502 write_store_listing(args, &store_path, SOURCE_CLIENT_STORE, "", load)
1503}
1504
1505fn write_store_listing(
1512 args: &cli::PoolListArgs,
1513 store_path: &Path,
1514 source: &'static str,
1515 provenance: &str,
1516 load: BuyerLoad,
1517) -> anyhow::Result<()> {
1518 let BuyerLoad {
1519 mut pools,
1520 mut skipped,
1521 } = load;
1522 pools.sort_by_key(|p| p.pool_id);
1524 skipped.sort_unstable();
1525 write_skipped_pools(&mut std::io::stderr().lock(), &skipped)?;
1526
1527 let mut out = std::io::stdout().lock();
1528 if args.json {
1529 let view = PoolListJson {
1530 store: store_path.display().to_string(),
1531 source,
1532 pools: pools.iter().map(PoolJson::from).collect(),
1533 skipped: skipped.iter().map(|p| format!("{p:#x}")).collect(),
1534 };
1535 serde_json::to_writer_pretty(&mut out, &view)?;
1536 writeln!(out)?;
1537 } else {
1538 writeln!(out, "store={}{provenance}", store_path.display())?;
1539 write_pools(&mut out, &pools, &skipped)?;
1540 }
1541 Ok(())
1542}
1543
1544async fn list_from_daemon(
1563 args: &cli::PoolListArgs,
1564 config_path: Option<&Path>,
1565 data_dir: &Path,
1566) -> anyhow::Result<()> {
1567 let buyer_db = node_buyer_db(data_dir);
1568 anyhow::ensure!(
1569 args.timeout_ms > 0,
1570 "--timeout-ms must be > 0 (jsonrpsee treats Duration::ZERO as \
1571 'never' rather than 'sub-millisecond deadline')"
1572 );
1573 let url = crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path)?;
1574 let client = jsonrpsee::http_client::HttpClientBuilder::default()
1575 .request_timeout(std::time::Duration::from_millis(args.timeout_ms))
1576 .build(&url)
1577 .with_context(|| format!("failed to build admin JSON-RPC client for {url}"))?;
1578
1579 let resp: BuyerPoolsResponse = match client.pools().await {
1580 Ok(resp) => resp,
1581 Err(jsonrpsee::core::client::Error::Transport(inner))
1584 if crate::commands::node::is_connection_refused(inner.as_ref()) =>
1585 {
1586 return list_from_stopped_daemon(args, data_dir, &buyer_db, &url, inner.as_ref());
1587 }
1588 Err(err) => {
1589 return Err(unreachable_daemon_error(
1590 args, data_dir, &buyer_db, &url, err,
1591 ));
1592 }
1593 };
1594
1595 let mut out = std::io::stdout().lock();
1596 if args.json {
1597 let view = DaemonPoolListJson {
1598 store: buyer_db.display().to_string(),
1599 source: SOURCE_DAEMON,
1600 pools: &resp.pools,
1601 skipped: &resp.skipped,
1602 };
1603 serde_json::to_writer_pretty(&mut out, &view)?;
1604 writeln!(out)?;
1605 } else {
1606 write_skipped_buyer_pools(&mut std::io::stderr().lock(), &resp.skipped)?;
1607 writeln!(
1608 out,
1609 "store={} (read from the running daemon)",
1610 buyer_db.display()
1611 )?;
1612 write_buyer_pools(&mut out, &resp)?;
1613 }
1614 Ok(())
1615}
1616
1617fn unreachable_daemon_error(
1620 args: &cli::PoolListArgs,
1621 data_dir: &Path,
1622 buyer_db: &Path,
1623 url: &str,
1624 err: jsonrpsee::core::client::Error,
1625) -> anyhow::Error {
1626 {
1627 let answered = matches!(err, jsonrpsee::core::client::Error::Call(_));
1631 let classified = crate::commands::node::classify_client_error(url, args.timeout_ms, err);
1632 if answered {
1633 classified.context(format!(
1634 "{} is a decdn-node data dir, so its buyer pools live in {}, which this command \
1635 reads through the daemon — a running daemon holds that file exclusively. The \
1636 daemon answered and refused the read.",
1637 data_dir.display(),
1638 buyer_db.display(),
1639 ))
1640 } else {
1641 classified.context(format!(
1642 "{} is a decdn-node data dir, so the pools that matter are in {} — which a \
1643 running daemon holds exclusively and no other process can open. This command \
1644 therefore asks the daemon, and the daemon did not answer. Start decdn-node, or \
1645 pass --data-dir <client dir> to inspect a client store instead.",
1646 data_dir.display(),
1647 buyer_db.display(),
1648 ))
1649 }
1650 }
1651}
1652
1653fn list_from_stopped_daemon(
1666 args: &cli::PoolListArgs,
1667 data_dir: &Path,
1668 buyer_db: &Path,
1669 url: &str,
1670 transport: &(dyn std::error::Error + Send + Sync + 'static),
1671) -> anyhow::Result<()> {
1672 match ReadOnlyBuyerPoolStore::open_file(buyer_db) {
1673 Ok(reader) => {
1674 let load = reader.load_all()?;
1675 write_store_listing(
1676 args,
1677 buyer_db,
1678 SOURCE_NODE_STORE_OFFLINE,
1679 " (read from disk; no daemon running)",
1680 load,
1681 )
1682 }
1683 Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => anyhow::bail!(
1684 "admin at {url} refused the connection ({transport}), but a process holds {} \
1685 exclusively — so a decdn-node IS running against {}. The admin URL or admin_port \
1686 is wrong, not the node down. Pass --admin-url, or check admin_port in the node's \
1687 config.",
1688 buyer_db.display(),
1689 data_dir.display(),
1690 ),
1691 Err(decdn_incentive::StoreError::NeedsRepair { .. }) => anyhow::bail!(
1695 "admin at {url} refused the connection ({transport}), and {} was not shut down \
1696 cleanly — it needs a repair pass that only a writable open can run. Start \
1697 decdn-node once against {} to repair it, then read it again.",
1698 buyer_db.display(),
1699 data_dir.display(),
1700 ),
1701 Err(decdn_incentive::StoreError::Absent { .. }) => anyhow::bail!(
1704 "admin at {url} refused the connection ({transport}), and {} does not exist — this \
1705 node holds no buyer pools on disk. It has not opened one, or the store was deleted \
1706 to force re-adoption. Start decdn-node and read it again, or run `decdn pool list \
1707 --all` to see what this keystore owns on chain.",
1708 buyer_db.display(),
1709 ),
1710 Err(store_err) => Err(anyhow::anyhow!(
1711 "admin at {url} refused the connection ({transport}); is the node running, and is \
1712 admin_port configured correctly? Reading {} from disk instead did not work either: \
1713 {store_err}",
1714 buyer_db.display(),
1715 )
1716 .context(format!(
1717 "{} is a decdn-node data dir, so the pools that matter are in {} — not in a client \
1718 store. Pass --data-dir <client dir> to inspect a client store instead.",
1719 data_dir.display(),
1720 buyer_db.display(),
1721 ))),
1722 }
1723}
1724
1725struct ChainPoolRow {
1730 pool_id: PoolId,
1732 status: PaymentPool::Status,
1734 deposit: U256,
1736 total_redeemed: U256,
1738 dispute_deadline: u64,
1741 tracked: Option<bool>,
1744}
1745
1746pub(crate) const fn status_label(status: PaymentPool::Status) -> &'static str {
1749 match status {
1750 PaymentPool::Status::Open => "open",
1751 PaymentPool::Status::Closing => "closing",
1752 PaymentPool::Status::Closed => "closed",
1753 _ => "?",
1755 }
1756}
1757
1758fn reclaimable_label(row: &ChainPoolRow, now: u64) -> String {
1760 match plan_reclaim(row.status, row.dispute_deadline, now) {
1761 ReclaimPlan::Reclaim => "now".to_string(),
1762 ReclaimPlan::SkipOpen => "close it first".to_string(),
1763 ReclaimPlan::SkipInWindow(deadline) => format_expiry(deadline, now),
1764 ReclaimPlan::SkipClosed => "already reclaimed".to_string(),
1765 ReclaimPlan::SkipUnknown => "?".to_string(),
1766 }
1767}
1768
1769const fn tracked_label(tracked: Option<bool>) -> &'static str {
1774 match tracked {
1775 Some(true) => "yes",
1776 Some(false) => "no",
1777 None => "?",
1778 }
1779}
1780
1781fn write_chain_pools(
1787 w: &mut impl Write,
1788 owner: Address,
1789 chain_id: u64,
1790 rows: &[ChainPoolRow],
1791 now: u64,
1792 local_store_read: bool,
1793) -> std::io::Result<()> {
1794 writeln!(
1799 w,
1800 "owner={owner:#x} chain_id={chain_id} pools={} local_store={}",
1801 rows.len(),
1802 if local_store_read {
1803 "read"
1804 } else {
1805 "unreadable"
1806 },
1807 )?;
1808 if rows.is_empty() {
1809 writeln!(w, "(this keystore owns no pools on this contract)")?;
1810 return Ok(());
1811 }
1812 writeln!(
1813 w,
1814 "{:<14} {:<8} {:>14} {:>14} {:>7} RECLAIMABLE",
1815 "POOL", "STATUS", "DEPOSIT", "REDEEMED", "TRACKED"
1816 )?;
1817 for row in rows {
1818 writeln!(
1819 w,
1820 "{:<14} {:<8} {:>14} {:>14} {:>7} {}",
1821 short_hex(&format!("{:#x}", row.pool_id)),
1822 status_label(row.status),
1823 format_usdc_u256(row.deposit),
1824 format_usdc_u256(row.total_redeemed),
1825 tracked_label(row.tracked),
1826 reclaimable_label(row, now),
1827 )?;
1828 }
1829 Ok(())
1830}
1831
1832#[derive(Serialize)]
1838struct PoolListAllJson {
1839 source: &'static str,
1841 owner: String,
1843 chain_id: u64,
1845 payment_pool: String,
1847 local_store_read: bool,
1851 pools: Vec<ChainPoolJson>,
1853}
1854
1855#[derive(Serialize)]
1857struct ChainPoolJson {
1858 pool_id: String,
1860 status: &'static str,
1862 deposit_micro_usdc: String,
1864 total_redeemed_micro_usdc: String,
1866 dispute_deadline: u64,
1869 reclaimable_now: bool,
1871 tracked: Option<bool>,
1875}
1876
1877impl ChainPoolJson {
1878 fn at(row: &ChainPoolRow, now: u64) -> Self {
1881 Self {
1882 pool_id: format!("{:#x}", row.pool_id),
1883 status: status_label(row.status),
1884 deposit_micro_usdc: row.deposit.to_string(),
1885 total_redeemed_micro_usdc: row.total_redeemed.to_string(),
1886 dispute_deadline: row.dispute_deadline,
1887 reclaimable_now: matches!(
1888 plan_reclaim(row.status, row.dispute_deadline, now),
1889 ReclaimPlan::Reclaim
1890 ),
1891 tracked: row.tracked,
1892 }
1893 }
1894}
1895
1896const SOURCE_CLIENT_STORE: &str = "client_store";
1898const SOURCE_DAEMON: &str = "daemon";
1900const SOURCE_NODE_STORE_OFFLINE: &str = "node_store_offline";
1906const SOURCE_CHAIN: &str = "chain";
1908
1909fn write_skipped_pools(w: &mut impl Write, skipped: &[PoolId]) -> std::io::Result<()> {
1912 for pool_id in skipped {
1913 writeln!(
1914 w,
1915 "warning: buyer pool {pool_id:#x} could not be decoded; its deposit remains \
1916 escrowed but untracked until the record is repaired"
1917 )?;
1918 }
1919 Ok(())
1920}
1921
1922fn write_pools(
1925 w: &mut impl Write,
1926 pools: &[BuyerPoolState],
1927 skipped: &[PoolId],
1928) -> std::io::Result<()> {
1929 writeln!(w, "pools={}", pools.len())?;
1930 if pools.is_empty() {
1931 if skipped.is_empty() {
1932 writeln!(w, "(no tracked pools)")?;
1933 }
1934 return Ok(());
1935 }
1936 writeln!(
1937 w,
1938 "{:<14} {:<14} {:<14} {:>14} {:>8}",
1939 "POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
1940 )?;
1941 for p in pools {
1942 writeln!(
1943 w,
1944 "{:<14} {:<14} {:<14} {:>14} {:>8}",
1945 short_hex(&format!("{:#x}", p.pool_id)),
1946 short_hex(&format!("{:#x}", p.owner)),
1947 short_hex(&format!("{:#x}", p.token)),
1948 format_usdc_u256(p.deposit),
1949 p.lane_count(),
1950 )?;
1951 }
1952 Ok(())
1953}
1954
1955pub(crate) fn write_buyer_pools(
1970 w: &mut impl Write,
1971 resp: &BuyerPoolsResponse,
1972) -> std::io::Result<()> {
1973 writeln!(w, "pools={}", resp.pools.len())?;
1974 if resp.pools.is_empty() {
1975 if resp.skipped.is_empty() {
1976 writeln!(w, "(no tracked pools)")?;
1977 }
1978 return Ok(());
1979 }
1980 writeln!(
1981 w,
1982 "{:<14} {:<14} {:<14} {:>14} {:>8}",
1983 "POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
1984 )?;
1985 for p in &resp.pools {
1986 writeln!(
1987 w,
1988 "{:<14} {:<14} {:<14} {:>14} {:>8}",
1989 short_hex(&p.pool_id),
1990 short_hex(&p.owner),
1991 short_hex(&p.token),
1992 format_usdc_u256(U256::from(p.deposit_micro_usdc)),
1993 p.lanes.len(),
1994 )?;
1995 }
1996 Ok(())
1997}
1998
1999pub(crate) fn write_skipped_buyer_pools(
2002 w: &mut impl Write,
2003 skipped: &[String],
2004) -> std::io::Result<()> {
2005 for pool_id in skipped {
2006 writeln!(
2007 w,
2008 "warning: buyer pool {pool_id} could not be decoded; its deposit remains escrowed but \
2009 untracked until the record is repaired"
2010 )?;
2011 }
2012 Ok(())
2013}
2014
2015#[derive(Serialize)]
2021struct PoolListJson {
2022 store: String,
2023 source: &'static str,
2024 pools: Vec<PoolJson>,
2025 skipped: Vec<String>,
2026}
2027
2028#[derive(Serialize)]
2037struct DaemonPoolListJson<'a> {
2038 store: String,
2039 source: &'static str,
2040 pools: &'a [decdn_common::admin::BuyerPoolSnapshot],
2041 skipped: &'a [String],
2042}
2043
2044#[derive(Serialize)]
2048struct PoolJson {
2049 pool_id: String,
2050 chain_id: u64,
2051 payment_pool: String,
2052 owner: String,
2053 token: String,
2054 deposit_usdc: String,
2055 lanes: Vec<LaneJson>,
2056}
2057
2058#[derive(Serialize)]
2059struct LaneJson {
2060 signer: String,
2061 provider: String,
2062 last_amount_usdc: String,
2063 last_bytes_delivered: String,
2064}
2065
2066impl From<&BuyerPoolState> for PoolJson {
2067 fn from(p: &BuyerPoolState) -> Self {
2068 Self {
2069 pool_id: format!("{:#x}", p.pool_id),
2070 chain_id: p.deployment.chain_id,
2071 payment_pool: format!("{:#x}", p.deployment.payment_pool),
2072 owner: format!("{:#x}", p.owner),
2073 token: format!("{:#x}", p.token),
2074 deposit_usdc: format_usdc_u256(p.deposit),
2075 lanes: p
2076 .lanes()
2077 .map(|(lane, progress)| LaneJson {
2078 signer: format!("{:#x}", lane.signer),
2079 provider: format!("{:#x}", lane.provider),
2080 last_amount_usdc: format_usdc_u256(progress.last_amount),
2081 last_bytes_delivered: progress.last_bytes.to_string(),
2082 })
2083 .collect(),
2084 }
2085 }
2086}
2087
2088fn format_usdc_u256(micro: U256) -> String {
2091 let scale = U256::from(1_000_000u64);
2092 let whole = micro / scale;
2093 let frac = (micro % scale).to::<u64>();
2095 format!("{whole}.{frac:06}")
2096}
2097
2098fn short_hex(hex: &str) -> String {
2102 match hex.get(..12) {
2103 Some(prefix) if hex.len() > 12 => format!("{prefix}…"),
2104 _ => hex.to_string(),
2105 }
2106}
2107
2108#[cfg(test)]
2109#[allow(clippy::unwrap_used, clippy::expect_used)]
2110mod tests;