# ClientConnector Refactor Part 5: SocketConnector / ProxyConnector Split

## Overview

This refactor separates the `ClientConnector` trait into two distinct traits with clear responsibilities:

- **`SocketConnector`**: Creates socket connections (TCP or QUIC) with bind_interface support
- **`ProxyConnector`**: Wraps protocols on existing streams

This separation reflects the reality that every `ClientConfig` implicitly defines both:
- Socket-level config: `bind_interface`, `transport`, `tcp_settings`, `quic_settings`
- Protocol-level config: `protocol`, `address`

## Current Problems

### 1. Mixed Responsibilities in `ClientConnector`

The current `ClientConnector` trait has methods for both socket creation and protocol setup:

```rust
pub trait ClientConnector: Send + Sync + Debug {
    fn bind_interface(&self) -> &Option<String>;
    fn supports_udp_over_tcp(&self) -> bool;
    fn proxy_location(&self) -> Option<&NetLocation>;

    // Socket creation (hop 0 only)
    async fn connect_as_first_hop(...) -> std::io::Result<TcpClientSetupResult>;
    async fn connect_udp_as_first_hop(...) -> std::io::Result<TcpClientUdpSetupResult>;

    // Protocol setup (hop 1+)
    async fn setup_on_existing_stream(...) -> std::io::Result<TcpClientSetupResult>;
    async fn setup_udp_on_existing_stream(...) -> std::io::Result<TcpClientUdpSetupResult>;
}
```

### 2. `TcpClientConnector` Does Too Much

`TcpClientConnector` currently:
- Stores `bind_interface`
- Creates TCP sockets with `new_tcp_socket()`
- Creates and caches QUIC endpoints
- Wraps protocols via `TcpClientHandler`

The socket creation code is only used at hop 0, but the struct carries this config for all hops.

### 3. `bind_interface` in Wrong Place

Currently `bind_interface` is on `ClientConfig`, meaning you can set it on any hop. But it only has effect at hop 0. This is confusing and leads to wasted config.

### 4. Redundant `[direct, vless]` vs `[vless]`

Currently these are equivalent when `vless` has the same `bind_interface`:
```yaml
# Option 1: explicit direct first hop
client_chain:
  - protocol:
      type: direct
    bind_interface: eth0
  - address: proxy:443
    protocol:
      type: vless

# Option 2: bind_interface on vless
client_chain:
  - address: proxy:443
    bind_interface: eth0
    protocol:
      type: vless
```

### 5. Pool Pairing Problem (CRITICAL)

**The original design had a critical flaw:** separating socket connectors and proxy connectors into independent pools with independent round-robin selection.

Consider this scenario:
```yaml
hops:
  - pool:
    - address: vmess.example.com:443
      transport: tcp
      protocol: vmess
    - address: vless.example.com:443
      transport: quic
      protocol: vless
    - transport: tcp
      bind_interface: eth0
      protocol: direct
```

With independent pools:
- `socket_connectors`: [TCP_to_vmess, QUIC_to_vless, TCP_eth0_direct]
- `proxy_hops[0]`: [vmess_handler, vless_handler] (no entry for direct)

After round-robins, we could get:
- Socket[2] (direct TCP) + Proxy[0] (vmess) → **Wrong!** Trying to do vmess handshake with... what server?
- Socket[0] (TCP to vmess) + Proxy[1] (vless) → **Wrong!** Sending vless protocol to vmess server

**The fix:** Socket and proxy for each pool entry at hop 0 must be **paired** and selected together.

## Design

### Key Insight

Every `ClientConfig` with a non-direct protocol implicitly defines **both** a `SocketConnector` and a `ProxyConnector`:

```yaml
address: proxy:443
bind_interface: eth0
transport: quic
quic_settings:
  sni_hostname: proxy.example.com
protocol:
  type: vless
  user_id: ...
```

This defines:
- **SocketConnector**: QUIC transport, bind to eth0, connect to proxy:443
- **ProxyConnector**: VLESS protocol targeting whatever the next hop is

When this config is used:
- **As hop 0**: Both SocketConnector and ProxyConnector are used
- **As hop 1+**: Only ProxyConnector is used, SocketConnector config is ignored

This means the same proxy definition can be reused in `client_groups` regardless of chain position.

### Hop 0 is Special: InitialHopEntry

Hop 0 is fundamentally different from subsequent hops:
- **Hop 0**: Creates socket AND optionally sets up protocol (if not direct)
- **Hops 1+**: Only set up protocol on existing stream

To handle mixed pools at hop 0, we use `InitialHopEntry`:

```rust
/// Entry in the initial hop (hop 0) pool.
///
/// Each entry pairs socket creation with optional protocol setup,
/// ensuring they are always selected together during round-robin.
pub enum InitialHopEntry {
    /// Direct connection - socket only, no protocol setup.
    /// Connects directly to the next hop's proxy or final destination.
    Direct(Box<dyn SocketConnector>),

    /// Proxy connection - socket + protocol setup.
    /// Socket connects to proxy_location, then protocol wraps the stream.
    Proxy {
        socket: Box<dyn SocketConnector>,
        proxy: Box<dyn ProxyConnector>,
    },
}
```

### New Traits

```rust
/// SocketConnector - Creates socket connections
///
/// Only used at hop 0. Handles TCP and QUIC transports with bind_interface.
#[async_trait]
pub trait SocketConnector: Send + Sync + Debug {
    /// Create a TCP/QUIC connection to the given address.
    async fn connect(
        &self,
        resolver: &Arc<dyn Resolver>,
        address: &NetLocation,
    ) -> std::io::Result<Box<dyn AsyncStream>>;

    /// Create UDP socket(s) for the given request type.
    /// Returns matched server/client stream pair.
    async fn connect_udp(
        &self,
        resolver: &Arc<dyn Resolver>,
        request: UdpStreamRequest,
    ) -> std::io::Result<TcpClientUdpSetupResult>;
}

/// ProxyConnector - Wraps protocols on existing streams
///
/// Used at all hops with non-direct protocols.
#[async_trait]
pub trait ProxyConnector: Send + Sync + Debug {
    /// The address of this proxy server (used to determine hop 0's connect target).
    fn proxy_location(&self) -> &NetLocation;

    /// Whether this protocol supports UDP-over-TCP tunneling.
    fn supports_udp_over_tcp(&self) -> bool;

    /// Setup protocol on existing stream, targeting `target`.
    async fn setup_tcp_stream(
        &self,
        stream: Box<dyn AsyncStream>,
        target: &NetLocation,
    ) -> std::io::Result<TcpClientSetupResult>;

    /// Setup UDP-over-TCP on existing stream.
    async fn setup_udp_stream(
        &self,
        stream: Box<dyn AsyncStream>,
        request: UdpStreamRequest,
    ) -> std::io::Result<TcpClientUdpSetupResult>;
}
```

### Implementation Structure

#### `SocketConnector` Implementation

```rust
/// Socket connector handling TCP and QUIC transports.
#[derive(Debug)]
pub struct SocketConnectorImpl {
    bind_interface: Option<String>,
    transport: TransportConfig,
}

#[derive(Debug)]
enum TransportConfig {
    Tcp {
        no_delay: bool,
    },
    Quic {
        endpoints: Vec<Arc<quinn::Endpoint>>,
        next_endpoint_index: AtomicU8,
        sni_hostname: Option<String>,
    },
}

impl SocketConnectorImpl {
    /// Create from ClientConfig's socket-related fields.
    ///
    /// # Arguments
    /// * `config` - The client config
    /// * `target_address` - Target for QUIC (IPv6 detection, default SNI).
    ///                      Pass None for direct protocol (forces TCP).
    pub fn from_config(config: &ClientConfig, target_address: Option<&NetLocation>) -> Option<Self> {
        // Direct protocol forces TCP (no QUIC for direct)
        let effective_transport = if config.protocol.is_direct() {
            Transport::Tcp
        } else {
            config.transport
        };

        let transport = match effective_transport {
            Transport::Tcp | Transport::Udp => {
                let no_delay = config.tcp_settings
                    .as_ref()
                    .map(|tc| tc.no_delay)
                    .unwrap_or(true);
                TransportConfig::Tcp { no_delay }
            }
            Transport::Quic => {
                let target_address = target_address
                    .expect("QUIC transport requires target_address");
                // Create and cache QUIC endpoints...
            }
        };

        Some(Self {
            bind_interface: config.bind_interface.clone().into_option(),
            transport,
        })
    }
}
```

#### `ProxyConnector` Implementation

```rust
/// Proxy connector for protocol wrapping.
#[derive(Debug)]
pub struct ProxyConnectorImpl {
    location: NetLocation,
    handler: Box<dyn TcpClientHandler>,
}

impl ProxyConnectorImpl {
    /// Create from the protocol-related fields of ClientConfig.
    /// Returns None for direct protocol.
    pub fn from_config(config: ClientConfig) -> Option<Self> {
        if config.protocol.is_direct() {
            return None;
        }

        let default_sni = config.address.address().hostname().map(ToString::to_string);
        let handler = create_tcp_client_handler(config.protocol, default_sni);

        Some(Self {
            location: config.address,
            handler,
        })
    }
}
```

### Updated `ClientProxyChain`

```rust
/// Entry in the initial hop (hop 0) pool.
pub enum InitialHopEntry {
    /// Direct connection - socket only, no protocol setup.
    Direct(Box<dyn SocketConnector>),

    /// Proxy connection - socket + protocol setup paired together.
    Proxy {
        socket: Box<dyn SocketConnector>,
        proxy: Box<dyn ProxyConnector>,
    },
}

pub struct ClientProxyChain {
    /// Hop 0 pool: paired socket + optional proxy.
    /// Each entry is selected atomically (socket and proxy together).
    initial_hop: Vec<InitialHopEntry>,
    initial_hop_next_index: AtomicU32,

    /// Hops 1+: just proxy connectors (no socket creation needed).
    /// These are selected independently per hop.
    subsequent_hops: Vec<Vec<Box<dyn ProxyConnector>>>,
    subsequent_next_indices: Vec<AtomicU32>,

    /// UDP-capable indices in initial_hop pool.
    udp_initial_hop_indices: Vec<usize>,
    udp_initial_hop_next_index: AtomicU32,
}

impl ClientProxyChain {
    pub fn new(
        initial_hop: Vec<InitialHopEntry>,
        subsequent_hops: Vec<Vec<Box<dyn ProxyConnector>>>,
    ) -> Self {
        assert!(!initial_hop.is_empty(), "must have at least one initial hop entry");

        // Compute UDP-capable indices
        let udp_initial_hop_indices: Vec<usize> = initial_hop
            .iter()
            .enumerate()
            .filter(|(_, entry)| {
                let entry_supports_udp = match entry {
                    InitialHopEntry::Direct(_) => true,
                    InitialHopEntry::Proxy { proxy, .. } => proxy.supports_udp_over_tcp(),
                };
                // Entry supports UDP if it supports UDP AND all subsequent hops support UDP
                entry_supports_udp && subsequent_hops.iter().all(|hop| {
                    hop.iter().any(|p| p.supports_udp_over_tcp())
                })
            })
            .map(|(i, _)| i)
            .collect();

        Self {
            initial_hop,
            initial_hop_next_index: AtomicU32::new(0),
            subsequent_hops,
            subsequent_next_indices: subsequent_hops.iter().map(|_| AtomicU32::new(0)).collect(),
            udp_initial_hop_indices,
            udp_initial_hop_next_index: AtomicU32::new(0),
        }
    }

    pub async fn connect_tcp(
        &self,
        remote_location: NetLocation,
        resolver: &Arc<dyn Resolver>,
    ) -> std::io::Result<TcpClientSetupResult> {
        // Select initial hop entry (round-robin)
        let entry = self.select_initial_hop_entry();

        // Select proxy connectors for subsequent hops
        let subsequent_proxies = self.select_subsequent_proxies();

        // Determine first target after initial hop
        let first_subsequent_target = subsequent_proxies
            .first()
            .map(|p| p.proxy_location())
            .unwrap_or(&remote_location);

        // Connect based on initial hop type
        let mut result = match entry {
            InitialHopEntry::Direct(socket) => {
                // Socket connects to first subsequent proxy (or final target)
                let stream = socket.connect(resolver, first_subsequent_target).await?;
                TcpClientSetupResult {
                    client_stream: stream,
                    early_data: None,
                }
            }
            InitialHopEntry::Proxy { socket, proxy } => {
                // Socket connects to this proxy
                let stream = socket.connect(resolver, proxy.proxy_location()).await?;
                // Protocol setup targeting first subsequent proxy (or final target)
                proxy.setup_tcp_stream(stream, first_subsequent_target).await?
            }
        };

        // Process subsequent hops
        for (i, proxy) in subsequent_proxies.iter().enumerate() {
            let target = subsequent_proxies
                .get(i + 1)
                .map(|p| p.proxy_location())
                .unwrap_or(&remote_location);

            result = proxy.setup_tcp_stream(result.client_stream, target).await?;
        }

        Ok(result)
    }

    pub async fn connect_udp(
        &self,
        resolver: &Arc<dyn Resolver>,
        request: UdpStreamRequest,
    ) -> std::io::Result<TcpClientUdpSetupResult> {
        // Select UDP-capable initial hop entry
        let entry = self.select_initial_hop_entry_for_udp()?;
        let subsequent_proxies = self.select_subsequent_proxies_for_udp()?;

        match entry {
            InitialHopEntry::Direct(socket) => {
                if subsequent_proxies.is_empty() {
                    // Direct UDP connection
                    socket.connect_udp(resolver, request).await
                } else {
                    // Direct socket + UDP-over-TCP through subsequent hops
                    let first_proxy = &subsequent_proxies[0];
                    let mut stream = socket.connect(resolver, first_proxy.proxy_location()).await?;

                    // Setup intermediate hops
                    for i in 0..subsequent_proxies.len() - 1 {
                        let target = subsequent_proxies[i + 1].proxy_location();
                        let result = subsequent_proxies[i].setup_tcp_stream(stream, target).await?;
                        stream = result.client_stream;
                    }

                    // Final hop: UDP stream
                    subsequent_proxies.last().unwrap()
                        .setup_udp_stream(stream, request).await
                }
            }
            InitialHopEntry::Proxy { socket, proxy } => {
                // Connect to initial proxy
                let mut stream = socket.connect(resolver, proxy.proxy_location()).await?;

                if subsequent_proxies.is_empty() {
                    // Initial proxy is final hop: UDP stream
                    proxy.setup_udp_stream(stream, request).await
                } else {
                    // Setup initial hop targeting first subsequent
                    let first_subsequent = subsequent_proxies[0].proxy_location();
                    let result = proxy.setup_tcp_stream(stream, first_subsequent).await?;
                    stream = result.client_stream;

                    // Setup intermediate subsequent hops
                    for i in 0..subsequent_proxies.len() - 1 {
                        let target = subsequent_proxies[i + 1].proxy_location();
                        let result = subsequent_proxies[i].setup_tcp_stream(stream, target).await?;
                        stream = result.client_stream;
                    }

                    // Final subsequent hop: UDP stream
                    subsequent_proxies.last().unwrap()
                        .setup_udp_stream(stream, request).await
                }
            }
        }
    }

    fn select_initial_hop_entry(&self) -> &InitialHopEntry {
        if self.initial_hop.len() == 1 {
            &self.initial_hop[0]
        } else {
            let idx = self.initial_hop_next_index.fetch_add(1, Ordering::Relaxed) as usize;
            &self.initial_hop[idx % self.initial_hop.len()]
        }
    }

    fn select_initial_hop_entry_for_udp(&self) -> std::io::Result<&InitialHopEntry> {
        if self.udp_initial_hop_indices.is_empty() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "No initial hop entries support UDP",
            ));
        }
        let idx = self.udp_initial_hop_next_index.fetch_add(1, Ordering::Relaxed) as usize;
        let pool_idx = self.udp_initial_hop_indices[idx % self.udp_initial_hop_indices.len()];
        Ok(&self.initial_hop[pool_idx])
    }
}
```

### Building from Config

```rust
fn build_client_proxy_chain(
    client_chain: OneOrSome<ClientChainHop>,
) -> ClientProxyChain {
    let hops: Vec<Vec<ClientConfig>> = /* parse hops into pools */;

    // Build initial hop entries (hop 0)
    let initial_hop: Vec<InitialHopEntry> = hops[0]
        .iter()
        .map(|config| {
            let socket = SocketConnectorImpl::from_config(config, Some(&config.address))
                .map(|s| Box::new(s) as Box<dyn SocketConnector>)
                .expect("Failed to create SocketConnector");

            match ProxyConnectorImpl::from_config(config.clone()) {
                Some(proxy) => InitialHopEntry::Proxy {
                    socket,
                    proxy: Box::new(proxy),
                },
                None => InitialHopEntry::Direct(socket),
            }
        })
        .collect();

    // Build subsequent hops (hops 1+)
    let subsequent_hops: Vec<Vec<Box<dyn ProxyConnector>>> = hops
        .into_iter()
        .skip(1)  // Skip hop 0
        .enumerate()
        .map(|(hop_index, hop_configs)| {
            hop_configs
                .into_iter()
                .map(|config| {
                    ProxyConnectorImpl::from_config(config.clone())
                        .map(|p| Box::new(p) as Box<dyn ProxyConnector>)
                        .unwrap_or_else(|| {
                            panic!(
                                "protocol: direct is only valid at hop 0. Found at hop {}",
                                hop_index + 1
                            )
                        })
                })
                .collect()
        })
        .collect();

    ClientProxyChain::new(initial_hop, subsequent_hops)
}
```

## Example Scenarios

### Scenario 1: Direct Connection with Bind Interface

```yaml
client_chain:
  - protocol:
      type: direct
    bind_interface: eth0
```

- `initial_hop`: [Direct(TCP socket, bind eth0)]
- `subsequent_hops`: []

Flow: Socket connects directly to remote_location.

### Scenario 2: Single Proxy Hop

```yaml
client_chain:
  - address: proxy:443
    bind_interface: eth0
    transport: quic
    protocol:
      type: vless
      user_id: ...
```

- `initial_hop`: [Proxy { socket: QUIC_eth0, proxy: VLESS }]
- `subsequent_hops`: []

Flow:
1. QUIC socket connects to proxy:443
2. VLESS wraps stream targeting remote_location

### Scenario 3: Multi-Hop Chain

```yaml
client_chain:
  - protocol:
      type: direct
    bind_interface: eth0
  - address: proxy1:443
    protocol:
      type: tls
      protocol:
        type: vless
        user_id: ...
  - address: proxy2:443
    protocol:
      type: vmess
      cipher: aes-128-gcm
      user_id: ...
```

- `initial_hop`: [Direct(TCP socket, bind eth0)]
- `subsequent_hops`: [[TLS+VLESS], [VMess]]

Flow:
1. TCP socket (eth0) connects to proxy1:443
2. TLS+VLESS wraps stream targeting proxy2:443
3. VMess wraps stream targeting remote_location

### Scenario 4: Mixed Pool at Hop 0 (THE CRITICAL FIX)

```yaml
client_chain:
  - pool:
    - address: vmess.example.com:443
      transport: tcp
      protocol: vmess
    - address: vless.example.com:443
      transport: quic
      protocol: vless
    - transport: tcp
      bind_interface: eth0
      protocol: direct
```

- `initial_hop`: [
    Proxy { socket: TCP, proxy: VMess(vmess.example.com) },
    Proxy { socket: QUIC, proxy: VLESS(vless.example.com) },
    Direct(TCP socket, bind eth0)
  ]
- `subsequent_hops`: []

Flow (round-robin selection picks the **paired** entry):
- Entry 0: TCP socket connects to vmess.example.com, VMess protocol
- Entry 1: QUIC socket connects to vless.example.com, VLESS protocol
- Entry 2: TCP socket (eth0) connects directly to remote_location

**Each socket is always paired with its correct protocol!**

### Scenario 5: Load-Balanced Interfaces

```yaml
client_chain:
  - pool:
    - protocol:
        type: direct
      bind_interface: eth0
    - protocol:
        type: direct
      bind_interface: eth1
    - protocol:
        type: direct
      bind_interface: eth2
  - address: proxy:443
    protocol:
      type: vless
      user_id: ...
```

- `initial_hop`: [Direct(eth0), Direct(eth1), Direct(eth2)]
- `subsequent_hops`: [[VLESS]]

Flow: Round-robin across direct entries, all then go through same VLESS proxy.

### Scenario 6: Reusable Proxy Definition

```yaml
client_groups:
  my-vless-quic:
    address: proxy:443
    bind_interface: eth0
    transport: quic
    protocol:
      type: vless
      user_id: ...

# Used as hop 0 - both socket and proxy config used
client_chain:
  - group: my-vless-quic

# Used as hop 1 - only proxy config used (socket config ignored)
client_chain:
  - protocol:
      type: direct
    bind_interface: eth1
  - group: my-vless-quic
```

In the second case, the QUIC transport and eth0 bind_interface from `my-vless-quic` are ignored - the connection uses eth1 TCP from the explicit direct hop.

## Cleanup Summary

### Files to Modify

1. **`src/tcp/client_connector.rs`** → Split into `socket_connector.rs` and `proxy_connector.rs`
   - Remove old `ClientConnector` trait
   - Add `SocketConnector` trait
   - Add `ProxyConnector` trait

2. **`src/tcp/local_connector.rs`** → Rename to `socket_connector_impl.rs`
   - Implement `SocketConnector` (not `ClientConnector`)
   - Add QUIC transport support (moved from `TcpClientConnector`)
   - Remove `ClientConnector` impl

3. **`src/tcp/tcp_client_connector.rs`** → Rename to `proxy_connector_impl.rs`
   - Remove `bind_interface` field
   - Remove `create_transport()` method
   - Remove `TransportConfig` enum
   - Remove `connect_as_first_hop` / `connect_udp_as_first_hop`
   - Implement only `ProxyConnector` trait

4. **`src/client_proxy_chain.rs`**
   - Add `InitialHopEntry` enum
   - Change structure to `initial_hop: Vec<InitialHopEntry>` and `subsequent_hops`
   - Update `connect_tcp` / `connect_udp` methods
   - Update selection logic to pair socket+proxy for hop 0

5. **`src/tcp/chain_builder.rs`**
   - Update `build_client_proxy_chain` to create `InitialHopEntry` values
   - Build paired entries for hop 0
   - Build proxy-only entries for subsequent hops

### Code Removed

- `TcpClientConnector::bind_interface` field
- `TcpClientConnector::create_transport()` method
- `TcpClientConnector::connect_as_first_hop()`
- `TcpClientConnector::connect_udp_as_first_hop()`
- `ClientConnector::bind_interface()` method
- `ClientConnector::connect_as_first_hop()` method
- `ClientConnector::connect_udp_as_first_hop()` method
- Duplicate TCP socket creation code (consolidated into `SocketConnectorImpl`)
- Duplicate QUIC endpoint code (consolidated into `SocketConnectorImpl`)
- `tcp_handler_util.rs` (split into `tcp_server_handler_factory.rs` and `tcp_client_handler_factory.rs`)

### Validation

Existing validation remains:
- `protocol: direct` at hop 1+ is rejected (already implemented)
- Socket config on hop 1+ is silently ignored (not an error - allows proxy reuse)

## Migration

This is an internal refactor with no config changes required. Existing configs continue to work:
- `bind_interface` on any hop still works (ignored at hop 1+)
- `transport: quic` on any hop still works (ignored at hop 1+)
- `protocol: direct` continues to mean "no protocol wrapping"

## Testing

1. All existing tests should pass unchanged
2. Add unit tests for `SocketConnector` trait
3. Add unit tests for `ProxyConnector` trait
4. Add unit tests for `InitialHopEntry` selection
5. Verify QUIC endpoint caching works correctly
6. Verify round-robin selection pairs socket+proxy correctly for hop 0
7. Test mixed pools at hop 0 (direct + various proxy types)
