Struct Storage

pub struct Storage { /* private fields */ }

Storage backend for TAP transactions and message audit trail

This struct provides the main interface for storing and retrieving TAP data from a SQLite database. It maintains two separate tables:

• transactions: For Transfer and Payment messages requiring business logic
• messages: For complete audit trail of all messages

It uses sqlx’s built-in connection pooling for efficient concurrent access and provides a native async API.

Example

use tap_node::storage::{Storage, MessageDirection};
use std::path::PathBuf;

// Create storage with default path
let storage = Storage::new(None).await?;

// Create storage with DID-based path
let agent_did = "did:web:example.com";
let storage_with_did = Storage::new_with_did(agent_did, None).await?;

// Create storage with custom TAP root
let custom_root = PathBuf::from("/custom/tap/root");
let storage_custom = Storage::new_with_did(agent_did, Some(custom_root)).await?;

// Query transactions
let transactions = storage.list_transactions(10, 0).await?;

// Query audit trail
let all_messages = storage.list_messages(20, 0, None).await?;
let incoming_only = storage.list_messages(10, 0, Some(MessageDirection::Incoming)).await?;

Implementations

impl Storage

pub async fn new_with_did( agent_did: &str, tap_root: Option<PathBuf>, ) -> Result<Self, StorageError>

Create a new Storage instance with an agent DID

This will initialize a SQLite database in the TAP directory structure:

• Default: ~/.tap/{did}/transactions.db
• Custom root: {tap_root}/{did}/transactions.db

Arguments

• agent_did - The DID of the agent this storage is for
• tap_root - Optional custom root directory (defaults to ~/.tap)

Errors

Returns StorageError if:

• Database initialization fails
• Migrations fail to run
• Connection pool cannot be created

pub async fn new(path: Option<PathBuf>) -> Result<Self, StorageError>

Create a new Storage instance

This will initialize a SQLite database at the specified path (or default location), run any pending migrations, and set up a connection pool.

Arguments

• path - Optional path to the database file. If None, uses TAP_NODE_DB_PATH env var or defaults to ./tap-node.db

Errors

Returns StorageError if:

• Database initialization fails
• Migrations fail to run
• Connection pool cannot be created

pub fn default_logs_dir(tap_root: Option<PathBuf>) -> PathBuf

Get the default logs directory

Returns the default directory for log files:

• Default: ~/.tap/logs
• Custom root: {tap_root}/logs

Arguments

• tap_root - Optional custom root directory (defaults to ~/.tap)

pub async fn update_message_status( &self, message_id: &str, status: &str, ) -> Result<(), StorageError>

Update the status of a message in the messages table

Arguments

• message_id - The ID of the message to update
• status - The new status (accepted, rejected, pending)

Errors

Returns StorageError if the database update fails

pub async fn update_transaction_status( &self, transaction_id: &str, status: &str, ) -> Result<(), StorageError>

Update the status of a transaction in the transactions table

Arguments

• transaction_id - The reference ID of the transaction to update
• status - The new status (pending, confirmed, failed, cancelled, reverted)

Errors

Returns StorageError if the database update fails

pub async fn get_transaction_by_id( &self, reference_id: &str, ) -> Result<Option<Transaction>, StorageError>

Get a transaction by its reference ID

Arguments

• reference_id - The reference ID of the transaction

Returns

• Ok(Some(Transaction)) if found
• Ok(None) if not found
• Err(StorageError) on database error

pub async fn get_transaction_by_thread_id( &self, thread_id: &str, ) -> Result<Option<Transaction>, StorageError>

Get a transaction by thread ID

Arguments

• thread_id - The thread ID to search for

Returns

• Ok(Some(Transaction)) if found
• Ok(None) if not found
• Err(StorageError) on database error

pub async fn is_agent_authorized_for_transaction( &self, transaction_id: &str, agent_did: &str, ) -> Result<bool, StorageError>

Check if an agent is authorized for a transaction

This checks the transaction_agents table to see if the given agent is associated with the transaction.

Arguments

• transaction_id - The reference ID of the transaction
• agent_did - The DID of the agent to check

Returns

• Ok(true) if the agent is authorized
• Ok(false) if the agent is not authorized or transaction doesn’t exist
• Err(StorageError) on database error

pub async fn insert_transaction_agent( &self, transaction_id: &str, agent_did: &str, agent_role: &str, ) -> Result<(), StorageError>

Insert a transaction agent

Arguments

• transaction_id - The reference ID of the transaction
• agent_did - The DID of the agent
• agent_role - The role of the agent (sender, receiver, compliance, other)

Returns

• Ok(()) on success
• Err(StorageError) on database error

pub async fn update_transaction_agent_status( &self, transaction_id: &str, agent_did: &str, status: &str, ) -> Result<(), StorageError>

Update transaction agent status

Arguments

• transaction_id - The reference ID of the transaction
• agent_did - The DID of the agent
• status - The new status (pending, authorized, rejected, cancelled)

Returns

• Ok(()) on success
• Err(StorageError) on database error

pub async fn get_transaction_agents( &self, transaction_id: &str, ) -> Result<Vec<(String, String, String)>, StorageError>

Get all agents for a transaction

Arguments

• transaction_id - The reference ID of the transaction

Returns

• Ok(Vec<(agent_did, agent_role, status)>) on success
• Err(StorageError) on database error

pub async fn are_all_agents_authorized( &self, transaction_id: &str, ) -> Result<bool, StorageError>

Check if all agents have authorized the transaction

Arguments

• transaction_id - The reference ID of the transaction

Returns

• Ok(true) if all agents have authorized
• Ok(false) if any agent hasn’t authorized or has rejected/cancelled
• Err(StorageError) on database error

pub async fn insert_transaction( &self, message: &PlainMessage, ) -> Result<(), StorageError>

Insert a new transaction from a TAP message

This method extracts transaction details from a Transfer or Payment message and stores them in the database with a ‘pending’ status.

Arguments

• message - The DIDComm PlainMessage containing a Transfer or Payment body

Errors

Returns StorageError if:

• The message is not a Transfer or Payment type
• Database insertion fails
• The transaction already exists (duplicate reference_id)

pub async fn list_transactions( &self, limit: u32, offset: u32, ) -> Result<Vec<Transaction>, StorageError>

List transactions with pagination

Retrieves transactions ordered by creation time (newest first).

Arguments

• limit - Maximum number of transactions to return
• offset - Number of transactions to skip (for pagination)

Returns

A vector of transactions ordered by creation time descending

pub async fn log_message( &self, message: &PlainMessage, direction: MessageDirection, raw_message: Option<&str>, ) -> Result<(), StorageError>

Log an incoming or outgoing message to the audit trail

This method stores any DIDComm message for audit purposes, regardless of type.

Arguments

• message - The DIDComm PlainMessage to log
• direction - Whether the message is incoming or outgoing
• raw_message - Optional raw JWE/JWS message string

Errors

Returns StorageError if:

• Database insertion fails
• The message already exists (duplicate message_id)

pub async fn get_message_by_id( &self, message_id: &str, ) -> Result<Option<Message>, StorageError>

Retrieve a message by its ID

Arguments

• message_id - The unique message ID

Returns

• Ok(Some(Message)) if found
• Ok(None) if not found
• Err(StorageError) on database error

pub async fn list_messages( &self, limit: u32, offset: u32, direction: Option<MessageDirection>, ) -> Result<Vec<Message>, StorageError>

List messages with pagination and optional filtering

Arguments

• limit - Maximum number of messages to return
• offset - Number of messages to skip (for pagination)
• direction - Optional filter by message direction

Returns

A vector of messages ordered by creation time descending

Trait Implementations

impl Clone for Storage

fn clone(&self) -> Storage

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.