1import asyncio
 2from collections import defaultdict
 3from typing import TYPE_CHECKING
 4
 5from app import marznode
 6from .grpcio import MarzNodeGRPCIO
 7from .grpclib import MarzNodeGRPCLIB
 8from ..models.node import NodeConnectionBackend
 9from ..models.user import User
10
11if TYPE_CHECKING:
12    from app.db import User as DBUser
13
14
15def update_user(
16    user: "DBUser", old_inbounds: set | None = None, remove: bool = False
17):
18    """updates a user on all related nodes"""
19    if old_inbounds is None:
20        old_inbounds = set()
21
22    node_inbounds = defaultdict(list)
23    if remove:
24        for inb in user.inbounds:
25            node_inbounds[inb.node_id]
26    else:
27        for inb in user.inbounds:
28            node_inbounds[inb.node_id].append(inb.tag)
29
30    for inb in old_inbounds:
31        node_inbounds[inb[0]]
32
33    for node_id, tags in node_inbounds.items():
34        if marznode.nodes.get(node_id):
35            asyncio.ensure_future(
36                marznode.nodes[node_id].update_user(
37                    user=User.model_validate(user), inbounds=tags
38                )
39            )
40
41
42async def remove_user(user: "DBUser"):
43    node_ids = set(inb.node_id for inb in user.inbounds)
44
45    for node_id in node_ids:
46        if marznode.nodes.get(node_id):
47            asyncio.ensure_future(
48                marznode.nodes[node_id].update_user(user=user, inbounds=[])
49            )
50
51
52async def remove_node(node_id: int):
53    if node_id in marznode.nodes:
54        await marznode.nodes[node_id].stop()
55        del marznode.nodes[node_id]
56
57
58async def add_node(db_node, certificate):
59    await remove_node(db_node.id)
60    if db_node.connection_backend == NodeConnectionBackend.grpcio:
61        node = MarzNodeGRPCIO(
62            db_node.id,
63            db_node.address,
64            db_node.port,
65            usage_coefficient=db_node.usage_coefficient,
66        )
67    else:
68        node = MarzNodeGRPCLIB(
69            db_node.id,
70            db_node.address,
71            db_node.port,
72            certificate.key,
73            certificate.certificate,
74            usage_coefficient=db_node.usage_coefficient,
75        )
76    marznode.nodes[db_node.id] = node
77
78
79__all__ = ["update_user", "add_node", "remove_node"]