ZerePy/src/actions/debridge_actions.py at bc658cfb726fb099e4638c2c3cb71441e3c5524e · blorm-network/ZerePy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import logging
from src.action_handler import register_action

logger = logging.getLogger("actions.debridge_actions")

@register_action("create-bridge-tx")
def create_bridge_tx(agent, **kwargs):
    """Create Bridge TX using Debridge"""
    try:
        required_args = ['connection', 'srcChainId', 'srcChainTokenIn', 'srcChainTokenInAmount', 'dstChainId', 'dstChainTokenOut', 'dstChainTokenOutRecipient']
        for arg in required_args:
            if arg not in kwargs:
                logger.error(f"Missing required argument: {arg}")
                return None

        response = agent.connection_manager.connections["debridge"].create_bridge_tx(
            connection=kwargs['connection'],
            srcChainId=kwargs['srcChainId'],
            srcChainTokenIn=kwargs['srcChainTokenIn'],
            srcChainTokenInAmount=kwargs['srcChainTokenInAmount'],
            dstChainId=kwargs['dstChainId'],
            dstChainTokenOut=kwargs['dstChainTokenOut'],
            dstChainTokenOutRecipient=kwargs['dstChainTokenOutRecipient'],
            dstChainTokenOutAmount=kwargs.get('dstChainTokenOutAmount', "auto"),
            dstChainOrderAuthorityAddress=kwargs.get('dstChainOrderAuthorityAddress'),
            affiliateFeeRecipient=kwargs.get('affiliateFeeRecipient'),
            prependOperatingExpense=kwargs.get('prependOperatingExpense', True),
            affiliateFeePercent=kwargs.get('affiliateFeePercent', 0),
            dlnHook=kwargs.get('dlnHook')
        )
        return response
    except Exception as e:
        logger.error(f"Failed to create bridge transaction: {str(e)}")
        return None

@register_action("get-order-status")
def get_order_status(agent, **kwargs):
    """Get order status using Debridge"""
    try:
        response = agent.connection_manager.connections["debridge"].get_order_status(
            id=kwargs.get('id'),
            hash=kwargs.get('hash')
        )
        return response
    except Exception as e:
        logger.error(f"Failed to get order status: {str(e)}")
        return None

@register_action("get-order-details")
def get_order_details(agent, **kwargs):
    """Get order details using Debridge"""
    try:
        if 'id' not in kwargs:
            logger.error("Missing required argument: id")
            return None

        response = agent.connection_manager.connections["debridge"].get_order_details(
            id=kwargs['id']
        )
        return response
    except Exception as e:
        logger.error(f"Failed to get order details: {str(e)}")
        return None

@register_action("cancel-tx")
def cancel_tx(agent, **kwargs):
    """Cancel transaction using Debridge"""
    try:
        if 'id' not in kwargs:
            logger.error("Missing required argument: id")
            return None

        response = agent.connection_manager.connections["debridge"].cancel_tx(
            id=kwargs['id']
        )
        return response
    except Exception as e:
        logger.error(f"Failed to cancel transaction: {str(e)}")
        return None

@register_action("extcall-cancel-tx")
def extcall_cancel_tx(agent, **kwargs):
    """External call to cancel transaction using Debridge"""
    try:
        if 'id' not in kwargs:
            logger.error("Missing required argument: id")
            return None

        response = agent.connection_manager.connections["debridge"].extcall_cancel_tx(
            id=kwargs['id']
        )
        return response
    except Exception as e:
        logger.error(f"Failed to perform external call to cancel transaction: {str(e)}")
        return None

@register_action("get-supported-chains")
def get_supported_chains(agent, **kwargs):
    """Get supported chains using Debridge"""
    try:
        response = agent.connection_manager.connections["debridge"].get_supported_chains()
        return response
    except Exception as e:
        logger.error(f"Failed to get supported chains: {str(e)}")
        return None

@register_action("get-token-list")
def get_token_list(agent, **kwargs):
    """Get token list using Debridge"""
    try:
        if 'chainId' not in kwargs:
            logger.error("Missing required argument: chainId")
            return None

        response = agent.connection_manager.connections["debridge"].get_token_list(
            chainId=kwargs['chainId']
        )
        return response
    except Exception as e:
        logger.error(f"Failed to get token list: {str(e)}")
        return None

@register_action("execute-bridge-tx")
def execute_bridge_tx(agent, **kwargs):
    """Execute bridge transaction using Debridge"""
    try:
        if 'connection' not in kwargs:
            logger.error("Missing required argument: connection")
            return None

        response = agent.connection_manager.connections["debridge"].execute_bridge_tx(
            connection=kwargs['connection'],
            compute_unit_price=kwargs.get('compute_unit_price', 200_000),
            compute_unit_limit=kwargs.get('compute_unit_limit')
        )
        return response
    except Exception as e:
        logger.error(f"Failed to execute bridge transaction: {str(e)}")
        return None