From Office-Bound Devices to Global Access: The BridgeLink Story

# ADB capabilities that changed everything:
adb devices                        # List all connected devices
adb -s SERIAL shell               # Execute shell commands
adb -s SERIAL install app.apk     # Install applications
adb -s SERIAL logcat              # Real-time log streaming
adb connect IP:PORT               # Connect to device over TCP/IP

# The revelation: ADB already supports remote connections!
adb connect 192.168.1.10:5555     # Local network
# But how do we make it work over internet? 🤔

# bore's beautiful simplicity:
bore local 5555 --to bore.pub
# Output: Listening at bore.pub:15750

# Now ADB can connect globally:
adb connect bore.pub:15750
# Result: Physical device accessible from anywhere! 🎉

#!/bin/bash
# Simple script to expose a device remotely

# Enable ADB TCP mode on USB-connected device
adb -s 1d752b81 tcpip 5555

# Forward local port to device
adb -s 1d752b81 forward tcp:5000 tcp:5555

# Create bore tunnel
bore local 5000 --to bore.pub &

# Output the connection URL
echo "Connect from anywhere:"
echo "adb connect bore.pub:15750"

# This quickly became unmaintainable
# tunnel_device_1.sh
# tunnel_device_2.sh
# tunnel_device_3.sh
# ...
# tunnel_device_50.sh 😱

# Problems:
# - No process management
# - No health monitoring
# - No automatic recovery
# - Port conflicts everywhere

# Breakthrough: Proper tunnel management with state tracking

class TunnelManager:
    def __init__(self):
        self.tunnels = {}  # session_id -> TunnelInfo
        self.state_file = Path.home() / ".bridgelink" / "tunnels.json"

    def create_tunnel(self, device_serial, adb_port, api_key, device_type):
        """Create and manage bore tunnel lifecycle"""

        # Start bore tunnel subprocess
        process = subprocess.Popen([
            'bore', 'local', str(adb_port),
            '--to', 'bridgelink.nativebridge.io',
            '--secret', api_key
        ], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        # Parse tunnel URL from bore output
        tunnel_url = self._parse_tunnel_url(process)

        # Store tunnel info
        tunnel_info = {
            'device_serial': device_serial,
            'url': tunnel_url,
            'pid': process.pid,
            'adb_port': adb_port,
            'device_type': device_type
        }

        # Persist state to survive restarts
        self._save_state()

        return tunnel_info

# BridgeLink's WiFi magic: 3-step automated setup

class WiFiConnectionFlow:
    def setup_wifi_device(self, usb_serial):
        """Transform USB device to WiFi device in 3 steps"""

        # Step 1: Enable TCP/IP mode on USB device
        subprocess.run(['adb', '-s', usb_serial, 'tcpip', '5555'])
        click.echo("✓ TCP/IP mode enabled on port 5555")

        # Step 2: Get device WiFi IP automatically
        result = subprocess.run(
            ['adb', '-s', usb_serial, 'shell', 'ip', 'route'],
            capture_output=True, text=True
        )
        device_ip = self._parse_ip_from_route(result.stdout)
        click.echo(f"✓ Device IP address: {device_ip}")

        # Step 3: Connect via WiFi
        subprocess.run(['adb', 'connect', f'{device_ip}:5555'])
        click.echo(f"✓ Connected via WiFi: {device_ip}:5555")

        # Now you can disconnect USB cable!
        click.echo("💡 You can now disconnect the USB cable!")
        click.echo("   The device will remain connected via WiFi.")

        return f"{device_ip}:5555"  # New WiFi serial

# Before WiFi support:
bridgelink devices add 1d752b81  # Must stay connected via USB

# After WiFi support:
bridgelink devices add --wifi 1d752b81
# Output:
# ✓ TCP/IP mode enabled
# ✓ Device IP: 192.168.1.15
# ✓ Connected via WiFi
# 💡 Disconnect USB cable now!

# Device is now wireless, tunnel still works globally! 🎉

class HealthMonitor:
    """Background daemon checking device health every 1 second"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.running = False
        self.check_interval = 1  # Fast 1-second checks

    async def monitor_devices(self):
        """Continuously check all registered devices"""

        while self.running:
            devices = self.api_client.list_devices()

            for device in devices:
                if device['device_state'] != 'active':
                    continue

                # Check if device is actually reachable
                is_healthy = await self._check_device_health(
                    device['device_serial'],
                    device['device_type']
                )

                if not is_healthy:
                    # Device is down! Auto-deactivate
                    logger.warning(f"Device {device['device_serial']} disconnected")

                    # Stop tunnel
                    self.tunnel_manager.stop_tunnel(device['device_serial'])

                    # Update backend state
                    self.api_client.update_device_state(
                        device['device_serial'],
                        'inactive'
                    )

            await asyncio.sleep(self.check_interval)

    async def _check_device_health(self, serial, device_type):
        """Smart health check based on device type"""

        if device_type == 'emulator':
            # Emulators: Check if process is running
            return self._check_emulator_running(serial)
        else:
            # Physical devices: Check ADB connectivity
            try:
                # Run off the event loop, so one slow device can't stall the others
                result = await asyncio.to_thread(
                    subprocess.run,
                    ['adb', '-s', serial, 'shell', 'echo', 'ping'],
                    capture_output=True,
                    timeout=5
                )
            except subprocess.TimeoutExpired:
                return False  # a hung device is an unhealthy device
            return result.returncode == 0

class ConnectionMonitor:
    """Auto-reconnect devices when they come back online"""

    async def watch_for_reconnections(self):
        """Monitor for devices coming back after disconnect"""

        while self.running:
            # Get all ADB-connected devices
            current_devices = set(ADBDeviceManager.list_devices())

            # Get devices with auto-activation enabled
            auto_activate_devices = self.api_client.get_auto_activate_devices()

            for device in auto_activate_devices:
                device_serial = device['device_serial']

                # Check if device just reconnected
                if device_serial in current_devices:
                    if device['device_state'] == 'inactive':
                        # Device is back! Auto-activate it
                        logger.info(f"Auto-activating {device_serial}")

                        # Create tunnel
                        is_wifi = self._is_wifi_connection(device_serial)
                        adb_port = self.tunnel_manager.setup_adb_tcp(
                            device_serial,
                            is_wifi=is_wifi
                        )
                        tunnel_info = self.tunnel_manager.create_tunnel(
                            device_serial, adb_port, self.api_key,
                            device['device_type']
                        )

                        # Update backend
                        device['device_details']['connection_mode'] = \
                            'WiFi' if is_wifi else 'USB'
                        self.api_client.add_device({
                            'device_serial': device_serial,
                            'device_type': device['device_type'],
                            'device_details': device['device_details'],
                            'tunnel_url': tunnel_info['url'],
                            'device_state': 'active'
                        })

                        logger.info(f"✅ {device_serial} auto-activated!")

            await asyncio.sleep(1)  # Check every second

# Developer unplugs device accidentally
# 1 second later: Health monitor detects, marks inactive

# Developer plugs device back in
# 1 second later: Connection monitor detects, auto-activates
# Tunnel recreated, backend updated, device ready!

# Total downtime: ~2 seconds
# Developer action required: Zero! 🎉

# Principle 1: Commands should be obvious
bridgelink devices add          # Not: bridgelink register-device
bridgelink devices list         # Not: bridgelink show-all-registered-devices

# Principle 2: Colorful and informative
✅ Authenticated as: you@example.com
📱 Fetching device information...
   Model: Galaxy S21
   Manufacturer: Samsung
   Android: 14 (SDK 34)
✓ Device 1d752b81 is now active!

# Principle 3: Guidance, not errors
❌ No devices found via ADB

Make sure:
  1. Device is connected via USB
  2. USB debugging is enabled
  3. ADB is installed and in PATH

# BridgeLink's color palette for maximum clarity

import click

# Success = Green
click.echo(click.style("✅ Device activated!", fg='green', bold=True))

# Errors = Red
click.echo(click.style("❌ Failed to connect", fg='red', bold=True))

# Warnings/Info = Yellow
click.echo(click.style("⚠️ Security Warning", fg='yellow', bold=True))

# Process Steps = Blue
click.echo(click.style("📡 Setting up WiFi...", fg='blue', bold=True))

# Values/URLs = Cyan/Magenta
click.echo(click.style("192.168.1.15:5555", fg='cyan', bold=True))
click.echo(click.style("bridgelink.io:15750", fg='magenta', bold=True))

# Complete CLI surface

@click.group()
def devices():
    """Manage Android devices"""
    pass

@devices.command(name='add')
@click.argument('device_serials', nargs=-1)
@click.option('--wifi', is_flag=True, help='Connect via WiFi')
@click.option('--auto-activate', is_flag=True, help='Auto-reconnect when plugged back in')
async def add_device(device_serials, wifi, auto_activate):
    """Register and activate devices with NativeBridge"""
    # Full WiFi setup flow
    # Tunnel creation
    # Backend registration
    # Health monitoring setup

@devices.command(name='list')
@click.option('--format', type=click.Choice(['table', 'json']))
def list_devices(format):
    """List all registered devices with beautiful table"""
    # Shows: #, Serial, Model, Brand, Type, Mode, State, Auto-Act, Tunnel URL

@devices.command(name='activate')
@click.argument('device_serial')
def activate_device(device_serial):
    """Reactivate existing device"""
    # Check backend registration
    # Create tunnel
    # Update state

@devices.command(name='deactivate')
@click.argument('device_serial', required=False)
@click.option('--all', is_flag=True)
def deactivate_device(device_serial, all):
    """Deactivate devices and stop tunnels"""
    # Stop tunnels
    # Update backend state

# Health Monitor - The Guardian
class HealthMonitorDaemon:
    """Runs in background, checks devices every 1 second"""

    def start(self, api_key):
        # Fork background process
        # Save PID to ~/.bridgelink/monitor.pid
        # Monitor all active devices
        # Auto-deactivate dead devices

# Connection Monitor - The Auto-Activator
class ConnectionMonitorDaemon:
    """Runs in background, watches for reconnected devices"""

    def start(self, api_key):
        # Fork background process
        # Save PID to ~/.bridgelink/connection_monitor.pid
        # Watch for auto-activate devices
        # Auto-create tunnels when device appears

# Enhanced device details collection
device_details = {
    'brand': 'Samsung',
    'model': 'Galaxy S21',
    'android_version': '14',
    'sdk_version': '34',
    'security_patch': '2024-10-01',
    'cpu': 'arm64-v8a',
    'ram_gb': '8.0',
    'storage_gb': '128',
    'resolution': '1080x2400',
    'density_dpi': '420',
    'connection_mode': 'WiFi'  # USB or WiFi
}

# Dashboard integration
dashboard_url = get_dashboard_url()
# dev: trust-me-bro.nativebridge.io/dashboard/bridgelink
# prod: nativebridge.io/dashboard/bridgelink

# Smart session detection
# If only 1 device: auto-select
# If multiple: show list to choose

class WiFiConnectionFlow:
    """Complete WiFi setup automation"""

    def setup_wifi_connection(self, usb_serial):
        """Transform USB device to WiFi device"""

        # Step 1: Enable TCP/IP mode
        click.echo(click.style("   Step 1/3: Enabling TCP/IP mode...", fg='blue'))
        result = subprocess.run(
            ['adb', '-s', usb_serial, 'tcpip', '5555'],
            capture_output=True, timeout=10
        )
        if result.returncode != 0:
            raise WiFiSetupError("Failed to enable TCP/IP mode")

        click.echo(click.style("   ✓ TCP/IP mode enabled on port 5555", fg='green'))
        time.sleep(3)  # Wait for mode switch

        # Step 2: Get device IP
        click.echo(click.style("   Step 2/3: Getting device IP address...", fg='blue'))
        result = subprocess.run(
            ['adb', '-s', usb_serial, 'shell', 'ip', 'route'],
            capture_output=True, text=True, timeout=10
        )

        device_ip = self._parse_ip_address(result.stdout)
        if not device_ip:
            raise WiFiSetupError("Could not determine device IP address")

        click.echo(click.style("   ✓ Device IP address: ", fg='green') +
                   click.style(device_ip, fg='cyan', bold=True))

        # Step 3: Connect via WiFi
        click.echo(click.style("   Step 3/3: Connecting via WiFi...", fg='blue'))
        result = subprocess.run(
            ['adb', 'connect', f'{device_ip}:5555'],
            capture_output=True, timeout=10
        )

        if result.returncode != 0:
            raise WiFiSetupError("Failed to connect via WiFi")

        click.echo(click.style(f"   ✓ Connected via WiFi: ", fg='green') +
                   click.style(f"{device_ip}:5555", fg='cyan', bold=True))

        # Success message
        click.echo(click.style("\n💡 You can now disconnect the USB cable!",
                              fg='yellow', bold=True))
        click.echo(click.style("   The device will remain connected via WiFi.",
                              fg='yellow'))

        return f"{device_ip}:5555"

class SmartAutoActivation:
    """Intelligent device reconnection"""

    async def handle_device_reconnection(self, device_serial):
        """Auto-activate device with full context awareness"""

        # Get device info from backend
        device = self.api_client.get_device(device_serial)

        if not device:
            logger.warning(f"Unknown device {device_serial} connected")
            return

        if not device.get('auto_activate'):
            logger.info(f"Device {device_serial} auto-activate disabled")
            return

        if device['device_state'] == 'active':
            logger.info(f"Device {device_serial} already active")
            return

        logger.info(f"🔄 Auto-activating {device_serial}...")

        # Detect connection mode
        is_wifi = self._is_wifi_connection(device_serial)
        connection_mode = 'WiFi' if is_wifi else 'USB'

        # Setup ADB TCP port forwarding
        adb_port = self.tunnel_manager.setup_adb_tcp(
            device_serial,
            is_wifi=is_wifi
        )

        if not adb_port:
            logger.error(f"Failed to setup ADB for {device_serial}")
            return

        # Create bore tunnel
        tunnel_info = self.tunnel_manager.create_tunnel(
            device_serial,
            adb_port,
            self.api_key,
            device['device_type']
        )

        if not tunnel_info:
            logger.error(f"Failed to create tunnel for {device_serial}")
            return

        # Update device details with current connection mode
        device_details = device['device_details']
        device_details['connection_mode'] = connection_mode

        # Update backend
        self.api_client.add_device({
            'device_serial': device_serial,
            'device_type': device['device_type'],
            'device_details': device_details,
            'tunnel_url': tunnel_info['url'],
            'device_state': 'active'
        })

        logger.info(f"✅ {device_serial} auto-activated!")
        logger.info(f"   Tunnel: {tunnel_info['url']}")
        logger.info(f"   Mode: {connection_mode}")

class DeviceInformationCollector:
    """Collect comprehensive device details via ADB"""

    def collect_device_info(self, serial):
        """Get all device information in one go"""

        # Basic info
        model = self._get_prop(serial, 'ro.product.model')
        manufacturer = self._get_prop(serial, 'ro.product.manufacturer')
        android_version = self._get_prop(serial, 'ro.build.version.release')
        sdk_version = self._get_prop(serial, 'ro.build.version.sdk')

        # Enhanced info
        security_patch = self._get_prop(serial, 'ro.build.version.security_patch')
        cpu = self._get_prop(serial, 'ro.product.cpu.abi')

        # RAM info
        ram_result = subprocess.run(
            ['adb', '-s', serial, 'shell', 'cat', '/proc/meminfo'],
            capture_output=True, text=True
        )
        ram_gb = self._parse_ram(ram_result.stdout)

        # Storage info
        storage_result = subprocess.run(
            ['adb', '-s', serial, 'shell', 'df', '/data'],
            capture_output=True, text=True
        )
        storage_gb = self._parse_storage(storage_result.stdout)

        # Screen info
        resolution = self._get_resolution(serial)
        density_dpi = self._get_density(serial)

        return DeviceInfo(
            serial=serial,
            model=model,
            android_version=android_version,
            sdk_version=sdk_version,
            manufacturer=manufacturer,
            security_patch=security_patch,
            cpu=cpu,
            ram_gb=ram_gb,
            storage_gb=storage_gb,
            resolution=resolution,
            density_dpi=density_dpi
        )

# Without auto-activation:
# Device disconnects → Manual: bridgelink devices activate SERIAL
# Device reconnects → Manual: bridgelink devices activate SERIAL

# With auto-activation:
# Device disconnects → Auto-deactivated (1 second detection)
# Device reconnects → Auto-activated (1 second detection)

# Developer action required: Zero! 🎉

# Before colors: Plain white text, hard to scan
# After colors: Success = green, errors = red, info = yellow

# Result: the one red line in a wall of output is the one you notice

# Challenge: bore output format varies by version
# Solution: Robust regex parsing with fallbacks

def parse_tunnel_url(self, bore_output):
    """Parse tunnel URL from various bore output formats"""

    patterns = [
        r'listening at ([a-z0-9\.\-]+:\d+)',  # v0.4.0+
        r'connected to ([a-z0-9\.\-]+:\d+)',  # v0.3.x
        r'tunnel to ([a-z0-9\.\-]+:\d+)'      # v0.2.x
    ]

    for pattern in patterns:
        match = re.search(pattern, bore_output, re.IGNORECASE)
        if match:
            return match.group(1)

    raise TunnelParseError("Could not parse tunnel URL")

# Challenge: Windows vs macOS/Linux path differences
# Solution: pathlib for everything

from pathlib import Path

# Works everywhere
STATE_DIR = Path.home() / ".bridgelink"
MONITOR_PID = STATE_DIR / "monitor.pid"
MONITOR_LOG = STATE_DIR / "monitor.log"

# Challenge: Zombie processes, orphaned tunnels
# Solution: Proper cleanup with signal handlers

import signal
import atexit

class TunnelManager:
    def __init__(self):
        self.tunnels = {}

        # Register cleanup handlers
        atexit.register(self.cleanup_all)
        signal.signal(signal.SIGTERM, self.signal_handler)
        signal.signal(signal.SIGINT, self.signal_handler)

    def cleanup_all(self):
        """Ensure all tunnels stopped on exit"""
        for serial in list(self.tunnels.keys()):
            self.stop_tunnel(serial)

# Attempted: Stream device screen like iOS Bridge
# Problem: a full streaming stack was out of scope (scrcpy already streams over ADB, no root needed)
# Decision: Focus on ADB access, let developers use scrcpy separately

# Attempted: Auto-install bore during pip install
# Problem: Platform-specific binaries, permission issues
# Solution: bridgelink setup command with clear instructions

# Attempted: Create multiple tunnels for same device
# Problem: ADB port conflicts, confusing for users
# Solution: One tunnel per device, clear ownership

# We especially need help with:
platform_support = [
    "Windows installer improvements",
    "Linux package managers (apt, yum)",
    "macOS Homebrew formula maintenance"
]

features = [
    "CI/CD integrations",
    "Team collaboration features",
    "Advanced device pooling"
]

documentation = [
    "Video tutorials",
    "Corporate deployment guides",
    "Troubleshooting database"
]

# Try it yourself:
pip install bridgelink   # or: brew tap AutoFlowLabs/tap && brew install bridgelink

# Quick setup:
bridgelink setup  # Installs ADB and bore

# Register your first device:
export NB_API_KEY="your-api-key"   # from the NativeBridge dashboard
bridgelink devices add

# View your devices:
bridgelink devices list

# Join the development:
git clone https://github.com/himanshkukreja/bridgelink
cd bridgelink
pip install -r requirements.txt
pip install -e .