hypercall/

directive_outbox.rs

use std::sync::Arc;
use std::time::Duration;

use tokio::sync::broadcast;
use tracing::{error, info};

use crate::shared::order_types::get_timestamp_millis;
pub use hypercall_db::{
    DirectiveDeliveryStatus, DirectiveDomainStatus, DirectiveOutboxAppend, DirectiveOutboxKind,
    DirectiveOutboxRow,
};
use hypercall_db::{DirectiveOutboxReader, DirectiveOutboxWriter};
use hypercall_signer::{RsmSigner, RsmSignerError, SignedDirective};
use hypercall_types::directives::ActionKey;
use hypercall_types::TransactionStatus;
use hypercall_types::{
    EngineMessage, SignedDirectiveTx, TransactionRequest, TransactionType, TransactionUpdate,
};

pub struct DirectiveDeliverySequencer {
    db: Arc<hypercall_db_diesel::DatabaseHandler>,
    rsm_signer: Arc<dyn RsmSigner>,
    transaction_request_sender: tokio::sync::mpsc::UnboundedSender<EngineMessage>,
    poll_interval: Duration,
}

impl DirectiveDeliverySequencer {
    pub fn new(
        db: Arc<hypercall_db_diesel::DatabaseHandler>,
        rsm_signer: Arc<dyn RsmSigner>,
        transaction_request_sender: tokio::sync::mpsc::UnboundedSender<EngineMessage>,
        poll_ms: u64,
    ) -> Self {
        Self {
            db,
            rsm_signer,
            transaction_request_sender,
            poll_interval: Duration::from_millis(poll_ms.max(1)),
        }
    }

    pub fn start_with_shutdown(
        self: Arc<Self>,
        mut shutdown_rx: broadcast::Receiver<()>,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            info!("Starting directive delivery sequencer");
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("Directive delivery sequencer received shutdown signal");
                        break;
                    }
                    _ = tokio::time::sleep(self.poll_interval) => {
                        if let Err(error) = self.drain_once().await {
                            error!(%error, "directive delivery sequencer drain failed");
                        }
                    }
                }
            }
            info!("Directive delivery sequencer stopped");
        })
    }

    async fn drain_once(&self) -> anyhow::Result<()> {
        let Some(row) = self.db.claim_next_directive_outbox_item_sync()? else {
            return Ok(());
        };

        if row.kind != DirectiveOutboxKind::NeedsRsmSignature.as_str() {
            let error = format!("unsupported directive outbox kind {}", row.kind);
            self.dead_letter_directive(&row, &error, DirectiveDeadLetterReason::UnsupportedKind)?;
            return Ok(());
        }

        // Use sign_preallocated with the nonce persisted in the outbox row.
        // The outbox append already allocated and persisted the nonce during
        // journaling, so calling sign() would allocate a second nonce and
        // desynchronize durable nonce state from directive metadata.
        let signed = match self
            .rsm_signer
            .sign_preallocated(&row.directive_id, &row.account, &row.payload, row.nonce)
            .await
        {
            Ok(signed) => signed,
            Err(error) => {
                if is_permanent_signing_error(&error) {
                    self.dead_letter_directive(
                        &row,
                        &error.to_string(),
                        DirectiveDeadLetterReason::from_signer_error(&error),
                    )?;
                } else {
                    self.db.mark_directive_outbox_delivery_failed_sync(
                        row.outbox_seq,
                        &error.to_string(),
                    )?;
                }
                return Err(anyhow::anyhow!(error.to_string()));
            }
        };

        match self.rsm_signer.is_nonce_used(signed.nonce).await {
            Ok(true) => {
                let error = nonce_consumed_manual_reconciliation_error(signed.nonce);
                tracing::warn!(
                    directive_id = %row.directive_id,
                    rsm_nonce = signed.nonce,
                    "RSM nonce already appears consumed on-chain before directive submission, stopping retries for manual reconciliation"
                );
                metrics::counter!(
                    "ht_directive_outbox_manual_reconciliation_required_total",
                    "action_key" => row.action_key.as_str(),
                    "reason_class" => "nonce_consumed",
                )
                .increment(1);
                emit_withdrawal_manual_reconciliation_signal(&row.directive_id, row.action_key);
                self.db
                    .mark_directive_outbox_manual_reconciliation_sync(row.outbox_seq, &error)?;
                return Ok(());
            }
            Ok(false) => {}
            Err(error) => {
                tracing::warn!(
                    directive_id = %row.directive_id,
                    %error,
                    "failed to check RSM nonce on-chain, proceeding with submission"
                );
            }
        }

        let tx_request = build_rsm_transaction_request(&row, signed);
        if self
            .transaction_request_sender
            .send(EngineMessage::TransactionRequest(tx_request))
            .is_err()
        {
            self.db.mark_directive_outbox_delivery_failed_sync(
                row.outbox_seq,
                "transaction request sender closed",
            )?;
            return Err(anyhow::anyhow!("transaction request sender closed"));
        }
        Ok(())
    }

    fn dead_letter_directive(
        &self,
        row: &DirectiveOutboxRow,
        error: &str,
        reason: DirectiveDeadLetterReason,
    ) -> anyhow::Result<()> {
        metrics::counter!(
            "ht_directive_outbox_dead_lettered_total",
            "action_key" => row.action_key.as_str(),
            "reason_class" => reason.as_str(),
        )
        .increment(1);
        emit_withdrawal_manual_reconciliation_signal(&row.directive_id, row.action_key);
        self.db
            .mark_directive_outbox_dead_lettered_sync(row.outbox_seq, error)?;
        let Some(update) = dead_lettered_withdrawal_update(row, error, get_timestamp_millis())
        else {
            return Ok(());
        };
        if self
            .transaction_request_sender
            .send(EngineMessage::TransactionUpdate(update))
            .is_err()
        {
            return Err(anyhow::anyhow!(
                "transaction update sender closed while dead-lettering directive {}",
                row.directive_id
            ));
        }
        Ok(())
    }
}

fn is_permanent_signing_error(error: &RsmSignerError) -> bool {
    matches!(
        error,
        RsmSignerError::InvalidAction(_)
            | RsmSignerError::UnsupportedAction(_)
            | RsmSignerError::IdempotencyConflict(_)
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DirectiveDeadLetterReason {
    UnsupportedKind,
    InvalidAction,
    UnsupportedAction,
    IdempotencyConflict,
}

impl DirectiveDeadLetterReason {
    fn from_signer_error(error: &RsmSignerError) -> Self {
        match error {
            RsmSignerError::InvalidAction(_) => Self::InvalidAction,
            RsmSignerError::UnsupportedAction(_) => Self::UnsupportedAction,
            RsmSignerError::IdempotencyConflict(_) => Self::IdempotencyConflict,
            RsmSignerError::SigningFailed(_) | RsmSignerError::PersistenceFailed(_) => {
                panic!("non-permanent signer error cannot be dead-lettered: {error}")
            }
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            Self::UnsupportedKind => "unsupported_kind",
            Self::InvalidAction => "invalid_action",
            Self::UnsupportedAction => "unsupported_action",
            Self::IdempotencyConflict => "idempotency_conflict",
        }
    }
}

fn nonce_consumed_manual_reconciliation_error(nonce: u64) -> String {
    format!(
        "RSM nonce {nonce} is already used before directive submission; on-chain outcome is unknown and requires manual reconciliation"
    )
}

fn emit_withdrawal_manual_reconciliation_signal(directive_id: &str, action_key: ActionKey) {
    if !matches!(
        action_key,
        ActionKey::SystemWithdrawToken | ActionKey::SystemCreditOption
    ) {
        return;
    }

    metrics::counter!(
        "ht_withdrawal_manual_reconciliation_required_total",
        "action_key" => action_key.as_str(),
    )
    .increment(1);
    tracing::warn!(
        directive_id = %directive_id,
        action_key = ?action_key,
        "Withdrawal directive requires manual reconciliation; automatic refunds are disabled"
    );
}

fn dead_lettered_withdrawal_update(
    row: &DirectiveOutboxRow,
    error: &str,
    timestamp: u64,
) -> Option<TransactionUpdate> {
    if !matches!(
        row.action_key,
        ActionKey::SystemWithdrawToken | ActionKey::SystemCreditOption
    ) {
        return None;
    }

    Some(TransactionUpdate {
        request_id: row.directive_id.clone(),
        status: TransactionStatus::Failed,
        tx_hash: None,
        error: Some(error.to_string()),
        timestamp,
        gas_used: None,
        gas_price: None,
    })
}

fn build_rsm_transaction_request(
    row: &DirectiveOutboxRow,
    signed: SignedDirective,
) -> TransactionRequest {
    let timestamp = get_timestamp_millis();
    TransactionRequest {
        request_id: row.directive_id.clone(),
        wallet_address: row.account,
        account_contract: row.account,
        transaction_type: TransactionType::RsmDirective(SignedDirectiveTx {
            directive: signed.directive,
            signature: signed.signature,
        }),
        timestamp,
        // Outbox rows represent domain-accepted work that still needs a chain
        // effect. The original request expiry must not turn transient delivery
        // failures into terminal failures, so each submit attempt gets only a
        // fresh in-process guard.
        expires_at: timestamp.saturating_add(300_000),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hypercall_types::wallet_address::test_wallet;

    #[test]
    fn only_permanent_signing_errors_dead_letter_outbox_rows() {
        assert!(is_permanent_signing_error(&RsmSignerError::InvalidAction(
            "malformed action".to_string()
        )));
        assert!(is_permanent_signing_error(
            &RsmSignerError::UnsupportedAction("unknown action".to_string())
        ));
        assert!(is_permanent_signing_error(
            &RsmSignerError::IdempotencyConflict("different payload".to_string())
        ));
        assert!(!is_permanent_signing_error(&RsmSignerError::SigningFailed(
            "relayer unavailable".to_string()
        )));
        assert!(!is_permanent_signing_error(
            &RsmSignerError::PersistenceFailed("db unavailable".to_string())
        ));
    }

    #[test]
    fn dead_letter_reason_uses_signer_error_variants() {
        assert_eq!(
            DirectiveDeadLetterReason::from_signer_error(&RsmSignerError::InvalidAction(
                "malformed action".to_string()
            ))
            .as_str(),
            "invalid_action"
        );
        assert_eq!(
            DirectiveDeadLetterReason::from_signer_error(&RsmSignerError::UnsupportedAction(
                "unknown action".to_string()
            ))
            .as_str(),
            "unsupported_action"
        );
        assert_eq!(
            DirectiveDeadLetterReason::from_signer_error(&RsmSignerError::IdempotencyConflict(
                "different payload".to_string()
            ))
            .as_str(),
            "idempotency_conflict"
        );
    }

    fn test_outbox_row(action_key: ActionKey) -> DirectiveOutboxRow {
        DirectiveOutboxRow {
            outbox_seq: 1,
            directive_id: "directive-1".to_string(),
            kind: DirectiveOutboxKind::NeedsRsmSignature.as_str().to_string(),
            action_key,
            account: test_wallet(1),
            signer: test_wallet(2),
            nonce: 7,
            payload: vec![1, 2, 3],
            expires_at_ms: None,
        }
    }

    #[test]
    fn dead_lettered_withdrawals_emit_failed_transaction_update() {
        let cash = dead_lettered_withdrawal_update(
            &test_outbox_row(ActionKey::SystemWithdrawToken),
            "bad directive",
            123,
        )
        .expect("cash withdrawal should emit an update");
        assert_eq!(cash.request_id, "directive-1");
        assert_eq!(cash.status, TransactionStatus::Failed);
        assert_eq!(cash.error.as_deref(), Some("bad directive"));
        assert_eq!(cash.timestamp, 123);

        let option = dead_lettered_withdrawal_update(
            &test_outbox_row(ActionKey::SystemCreditOption),
            "bad option directive",
            456,
        )
        .expect("option withdrawal should emit an update");
        assert_eq!(option.status, TransactionStatus::Failed);
        assert_eq!(option.error.as_deref(), Some("bad option directive"));
        assert_eq!(option.timestamp, 456);
    }

    #[test]
    fn dead_lettered_non_withdrawals_do_not_emit_transaction_update() {
        assert!(dead_lettered_withdrawal_update(
            &test_outbox_row(ActionKey::RsmHlLimitOrder),
            "bad order",
            123,
        )
        .is_none());
    }

    #[test]
    fn consumed_nonce_error_requires_manual_reconciliation() {
        let error = nonce_consumed_manual_reconciliation_error(7);
        assert!(error.contains("already used"));
        assert!(error.contains("manual reconciliation"));
    }

    #[test]
    fn rsm_transaction_request_uses_fresh_delivery_deadline() {
        let mut row = test_outbox_row(ActionKey::SystemWithdrawToken);
        row.nonce = 10;
        row.expires_at_ms = Some(1);
        let signed = SignedDirective {
            request_id: row.directive_id.clone(),
            account: row.account,
            nonce: row.nonce,
            directive: vec![1, 2, 3],
            signature: "0x1234".to_string(),
        };

        let tx_request = build_rsm_transaction_request(&row, signed);

        assert!(
            tx_request.expires_at > tx_request.timestamp,
            "delivery request should have a fresh in-process deadline"
        );
        assert_ne!(
            tx_request.expires_at,
            row.expires_at_ms.unwrap(),
            "domain request expiry must not terminal-expire accepted chain delivery"
        );
    }
}