Build / Integration Guide / Deposit Monitoring

5. Deposit Monitoring

1. Generate Deposit Address

Pseudocode

FUNCTION generateDepositAddress(userId):
    // Step 1: Input validation
    IF (userId IS empty OR userId IS invalid):
        THROW "Invalid user ID"

    // Step 2: Generate unique payment identifier
    timestamp = getCurrentTimestamp()
    randomBytes = generateSecureRandom(8) // 8 bytes = 64 bits
    paymentId = "deposit-" + userId + "-" + timestamp + "-" + randomBytes

    // Step 3: Convert payment ID to bytes for gRPC
    paymentIdBytes = convertToBytes(paymentId, "UTF-8")

    // Step 4: Call wallet gRPC to get address with payment ID
    request = {
        payment_id: paymentIdBytes
    }

    TRY:
        response = walletClient.GetPaymentIdAddress(request)

        // Step 5: Extract address information
        address = response.one_sided_address_base58
        emojiAddress = response.one_sided_address_emoji

        // Step 6: Store in database for tracking
        depositRecord = {
            userId: userId,
            paymentId: paymentId,
            address: address,
            status: "pending",
            createdAt: timestamp
        }

        database.save(depositRecord)

        // Step 7: Return deposit instructions to user
        RETURN {
            address: address,
            paymentId: paymentId,
            emojiAddress: emojiAddress,
            instructions: "Send XTM to this address with payment ID: " + paymentId,
            qrCode: generateQRCode(address, paymentId)
        }

    CATCH (grpcError):
        LOG "Failed to generate address: " + grpcError.message
        THROW "Address generation failed"

grpcurl

PAYMENT_ID_B64=$(echo -n "deposit-user123-$(date +%s)" | base64)

grpcurl -plaintext 
  -d "{"payment_id": "$PAYMENT_ID_B64"}" 
  localhost:18143 
  tari.rpc.Wallet/GetPaymentIdAddress

Node.js

async function generateDepositAddress(userId, walletClient, db) {
  if (!userId) throw new Error('Invalid user ID');

  const timestamp = Date.now();
  const randomBytes = crypto.randomBytes(8).toString('hex');
  const paymentId = `deposit-${userId}-${timestamp}-${randomBytes}`;
  const paymentIdBytes = Buffer.from(paymentId, 'utf-8');

  try {
    const response = await walletClient.getPaymentIdAddress({ paymentId: paymentIdBytes });

    await db.deposits.create({
      userId,
      paymentId,
      address: response.oneSidedAddressBase58,
      status: 'pending',
      createdAt: timestamp,
    });

    return {
      address: response.oneSidedAddressBase58,
      emojiAddress: response.oneSidedAddressEmoji,
      paymentId,
      instructions: `Send XTM to this address with payment ID: ${paymentId}`,
    };
  } catch (grpcError) {
    throw new Error(`Address generation failed: ${grpcError.message}`);
  }
}

Python

def generate_deposit_address(user_id, wallet_client, db):
    if not user_id:
        raise ValueError("Invalid user ID")

    timestamp = int(time.time())
    random_bytes = secrets.token_hex(8)
    payment_id = f"deposit-{user_id}-{timestamp}-{random_bytes}"
    payment_id_bytes = payment_id.encode("utf-8")

    try:
        response = wallet_client.GetPaymentIdAddress(payment_id=payment_id_bytes)

        db.deposits.save({
            "user_id": user_id,
            "payment_id": payment_id,
            "address": response.one_sided_address_base58,
            "status": "pending",
            "created_at": timestamp,
        })

        return {
            "address": response.one_sided_address_base58,
            "emoji_address": response.one_sided_address_emoji,
            "payment_id": payment_id,
            "instructions": f"Send XTM to this address with payment ID: {payment_id}",
        }
    except grpc.RpcError as e:
        raise RuntimeError(f"Address generation failed: {e.details()}")

Rust

async fn generate_deposit_address(
    user_id: &str,
    wallet_client: &mut WalletClient,
) -> Result> {
    if user_id.is_empty() {
        return Err("Invalid user ID".into());
    }

    let timestamp = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
    let random_bytes = hex::encode(rand::random::<[u8; 8]>());
    let payment_id = format!("deposit-{}-{}-{}", user_id, timestamp, random_bytes);

    let request = GetPaymentIdAddressRequest {
        payment_id: payment_id.clone().into_bytes(),
    };

    let response = wallet_client.get_payment_id_address(request).await?;

    Ok(DepositInfo {
        address: response.one_sided_address_base58,
        payment_id,
        instructions: format!("Send XTM to this address with payment ID: {}", payment_id),
    })
}

PHP

function generateDepositAddress(string $userId, WalletClient $walletClient, PDO $db): array {
    if (empty($userId)) {
        throw new InvalidArgumentException('Invalid user ID');
    }

    $timestamp = time();
    $randomBytes = bin2hex(random_bytes(8));
    $paymentId = "deposit-{$userId}-{$timestamp}-{$randomBytes}";

    try {
        $response = $walletClient->GetPaymentIdAddress(['payment_id' => $paymentId])->wait();

        $stmt = $db->prepare(
            'INSERT INTO deposits (user_id, payment_id, address, status, created_at) VALUES (?, ?, ?, ?, ?)'
        );
        $stmt->execute([$userId, $paymentId, $response->getOneSidedAddressBase58(), 'pending', $timestamp]);

        return [
            'address' => $response->getOneSidedAddressBase58(),
            'payment_id' => $paymentId,
            'instructions' => "Send XTM to this address with payment ID: {$paymentId}",
        ];
    } catch (Exception $e) {
        throw new RuntimeException('Address generation failed: ' . $e->getMessage());
    }
}

2. Monitor for Incoming Transactions

Pseudocode

FUNCTION monitorIncomingTransactions():
    processedTransactions = loadProcessedTransactionsList()

    WHILE (service_is_running):
        TRY:
            // Method 1: Real-time streaming (preferred)
            IF (streaming_available):
                transactionStream = walletClient.StreamTransactionEvents()

                FOR EACH event IN transactionStream:
                    IF (event.type == "Mined" AND event.direction == "Inbound"):
                        processTransactionEvent(event)

            // Method 2: Polling (backup)
            ELSE:
                completedTransactions = walletClient.GetCompletedTransactions()

                FOR EACH transaction IN completedTransactions:
                    IF (shouldProcessTransaction(transaction)):
                        processTransaction(transaction)

        CATCH (connection_error):
            LOG "Connection lost, retrying in 5 seconds..."
            SLEEP(5)
            CONTINUE

        SLEEP(30) // Poll every 30 seconds if streaming not available

grpcurl

# Poll for completed transactions (streaming requires a persistent client)
grpcurl -plaintext localhost:18143 tari.rpc.Wallet/GetCompletedTransactions

# Check a specific transaction's confirmation status
grpcurl -plaintext -d '{"transaction_ids": ["PLACEHOLDER_tx_id"]}' 
  localhost:18143 tari.rpc.Wallet/GetTransactionInfo

Node.js

async function monitorIncomingTransactions(walletClient, db) {
  const processed = await db.processedTx.getIds();

  while (serviceIsRunning) {
    try {
      const stream = walletClient.streamTransactionEvents({});

      for await (const event of stream) {
        if (event.type === 'Mined' && event.direction === 'Inbound') {
          await processTransaction(event.transaction, db, processed);
        }
      }
    } catch (err) {
      console.error('Connection lost, retrying in 5 seconds...', err);
      await sleep(5000);
      continue;
    }
    await sleep(30000);
  }
}

async function processTransaction(tx, db, processed) {
  const paymentId = tx.paymentId?.toString('utf-8');
  if (!paymentId) return;

  const deposit = await db.deposits.findByPaymentId(paymentId);
  if (!deposit) return;

  const amountXTM = tx.amount / 1_000_000;
  await db.transaction(async (trx) => {
    await trx.users.incrementBalance(deposit.userId, amountXTM);
    await trx.deposits.markProcessed(deposit.id, tx.txId);
  });

  processed.add(tx.txId);
}

Python

def monitor_incoming_transactions(wallet_client, db):
    processed = db.processed_tx.load_ids()

    while service_is_running:
        try:
            for event in wallet_client.StreamTransactionEvents():
                if event.type == "Mined" and event.direction == "Inbound":
                    process_transaction(event.transaction, db, processed)
        except grpc.RpcError:
            logging.warning("Connection lost, retrying in 5 seconds...")
            time.sleep(5)
            continue
        time.sleep(30)


def process_transaction(tx, db, processed):
    payment_id = tx.payment_id.decode("utf-8") if tx.payment_id else None
    if not payment_id:
        return

    deposit = db.deposits.find_by_payment_id(payment_id)
    if not deposit:
        return

    amount_xtm = tx.amount / 1_000_000
    with db.transaction():
        db.users.increment_balance(deposit.user_id, amount_xtm)
        db.deposits.mark_processed(deposit.id, tx.tx_id)

    processed.add(tx.tx_id)

Rust

async fn monitor_incoming_transactions(
    wallet_client: &mut WalletClient,
    db: &Db,
) -> Result<(), Box> {
    let mut processed = db.processed_tx_ids().await?;

    loop {
        let mut stream = wallet_client.stream_transaction_events(()).await?.into_inner();

        while let Some(event) = stream.message().await? {
            if event.event_type == "Mined" && event.direction == "Inbound" {
                process_transaction(&event.transaction, db, &mut processed).await?;
            }
        }

        tokio::time::sleep(Duration::from_secs(30)).await;
    }
}

async fn process_transaction(
    tx: &Transaction,
    db: &Db,
    processed: &mut HashSet,
) -> Result<(), Box> {
    let payment_id = String::from_utf8(tx.payment_id.clone())?;
    let deposit = match db.find_deposit_by_payment_id(&payment_id).await? {
        Some(d) => d,
        None => return Ok(()),
    };

    let amount_xtm = tx.amount as f64 / 1_000_000.0;
    db.credit_user_and_mark_processed(&deposit.user_id, amount_xtm, &tx.tx_id).await?;
    processed.insert(tx.tx_id.clone());

    Ok(())
}

PHP

function monitorIncomingTransactions(WalletClient $walletClient, PDO $db): void {
    $processed = loadProcessedTransactionIds($db);

    while (serviceIsRunning()) {
        try {
            $completed = $walletClient->GetCompletedTransactions(new Empty())->wait();

            foreach ($completed->getTransactions() as $tx) {
                if ($tx->getDirection() === 'INBOUND' && !in_array($tx->getTxId(), $processed)) {
                    processTransaction($tx, $db, $processed);
                }
            }
        } catch (Exception $e) {
            error_log('Connection lost, retrying in 5 seconds: ' . $e->getMessage());
            sleep(5);
            continue;
        }
        sleep(30);
    }
}

function processTransaction($tx, PDO $db, array &$processed): void {
    $paymentId = $tx->getPaymentId();
    if (empty($paymentId)) {
        return;
    }

    $stmt = $db->prepare('SELECT * FROM deposits WHERE payment_id = ?');
    $stmt->execute([$paymentId]);
    $deposit = $stmt->fetch();
    if (!$deposit) {
        return;
    }

    $amountXTM = $tx->getAmount() / 1_000_000;

    $db->beginTransaction();
    $db->prepare('UPDATE users SET balance = balance + ? WHERE id = ?')
        ->execute([$amountXTM, $deposit['user_id']]);
    $db->prepare('UPDATE deposits SET status = ?, tx_id = ? WHERE id = ?')
        ->execute(['completed', $tx->getTxId(), $deposit['id']]);
    $db->commit();

    $processed[] = $tx->getTxId();
}