# Phoenix Channels - Real-time WebSocket API

STA Connector provides real-time communication through Phoenix Channels, enabling live updates for file processing, metrics, logs, health status, and admin notifications.

## Connection

### WebSocket Endpoint

```
ws://localhost:4000/socket/websocket
wss://your-domain.com/socket/websocket  # Production with TLS
```

### Connection Parameters

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `token` | string | Yes | Phoenix.Token for authentication |
| `vsn` | string | No | Protocol version (default: "2.0.0") |

## Authentication

Phoenix Channels use `Phoenix.Token` for secure authentication. Tokens are signed server-side and verified on each connection.

### Token Generation (Server-side Elixir)

```elixir
# In your authentication controller or context
defmodule STAConnectorWeb.Auth do
  @token_salt "user_socket"
  @max_age 86_400  # 24 hours

  def generate_socket_token(user_id) do
    Phoenix.Token.sign(STAConnectorWeb.Endpoint, @token_salt, user_id)
  end

  def verify_socket_token(token) do
    Phoenix.Token.verify(STAConnectorWeb.Endpoint, @token_salt, token, max_age: @max_age)
  end
end
```

### Socket Authentication (Server-side)

```elixir
# lib/sta_connector_web/channels/user_socket.ex
defmodule STAConnectorWeb.UserSocket do
  use Phoenix.Socket

  channel "files:*", STAConnectorWeb.FilesChannel
  channel "metrics:*", STAConnectorWeb.MetricsChannel
  channel "logs:*", STAConnectorWeb.LogsChannel
  channel "health:*", STAConnectorWeb.HealthChannel
  channel "admin:*", STAConnectorWeb.AdminChannel

  @impl true
  def connect(%{"token" => token}, socket, _connect_info) do
    case Phoenix.Token.verify(STAConnectorWeb.Endpoint, "user_socket", token, max_age: 86_400) do
      {:ok, user_id} ->
        {:ok, assign(socket, :user_id, user_id)}
      {:error, _reason} ->
        :error
    end
  end

  def connect(_params, _socket, _connect_info), do: :error

  @impl true
  def id(socket), do: "user_socket:#{socket.assigns.user_id}"
end
```

---

## Channels

### 1. files:lobby - File Events

Real-time file processing events including creation, updates, completion, and failures.

#### Events

| Event | Direction | Description |
|-------|-----------|-------------|
| `file:created` | Server -> Client | New file received for processing |
| `file:updated` | Server -> Client | File status or metadata updated |
| `file:completed` | Server -> Client | File processing completed successfully |
| `file:failed` | Server -> Client | File processing failed |
| `subscribe_file` | Client -> Server | Subscribe to specific file updates |
| `unsubscribe_file` | Client -> Server | Unsubscribe from file updates |

#### Payload Structures

**file:created**
```json
{
  "id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c",
  "filename": "ARQC001_20260131_001.txt",
  "system": "CCS",
  "file_type": "ARQC001",
  "size": 15240,
  "source_ispb": "00000000",
  "destination_ispb": "12345678",
  "direction": "inbound",
  "status": "pending",
  "created_at": "2026-01-31T14:30:00Z"
}
```

**file:updated**
```json
{
  "id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c",
  "status": "processing",
  "progress": 45,
  "updated_at": "2026-01-31T14:30:15Z"
}
```

**file:completed**
```json
{
  "id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c",
  "status": "completed",
  "processed_at": "2026-01-31T14:30:30Z",
  "processing_time_ms": 30000,
  "records_processed": 1250,
  "checksum": "sha256:abc123..."
}
```

**file:failed**
```json
{
  "id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c",
  "status": "failed",
  "error_code": "PARSE_ERROR",
  "error_message": "Invalid record format at line 42",
  "failed_at": "2026-01-31T14:30:25Z",
  "retry_count": 3,
  "max_retries": 5
}
```

---

### 2. metrics:lobby - Metrics Updates

Real-time metrics and statistics updates.

#### Events

| Event | Direction | Description |
|-------|-----------|-------------|
| `metrics:update` | Server -> Client | Periodic metrics snapshot |
| `metrics:threshold_alert` | Server -> Client | Metric exceeded threshold |
| `get_current` | Client -> Server | Request current metrics |

#### Payload Structures

**metrics:update**
```json
{
  "timestamp": "2026-01-31T14:30:00Z",
  "inbound": {
    "files_processed": 1542,
    "files_pending": 23,
    "files_failed": 12,
    "bytes_processed": 157286400,
    "avg_processing_time_ms": 2340
  },
  "outbound": {
    "files_sent": 876,
    "files_pending": 5,
    "files_failed": 3,
    "bytes_sent": 89456000
  },
  "sta_client": {
    "requests_total": 4521,
    "requests_success": 4498,
    "requests_failed": 23,
    "rate_limit_remaining": 87,
    "circuit_breaker_state": "closed"
  },
  "system": {
    "uptime_seconds": 86400,
    "memory_used_mb": 256,
    "goroutines": 42,
    "db_connections": 10
  }
}
```

**metrics:threshold_alert**
```json
{
  "metric": "inbound.files_pending",
  "current_value": 150,
  "threshold": 100,
  "severity": "warning",
  "message": "High pending file count detected",
  "triggered_at": "2026-01-31T14:30:00Z"
}
```

---

### 3. logs:lobby - Log Streaming

Real-time log streaming for monitoring and debugging.

#### Events

| Event | Direction | Description |
|-------|-----------|-------------|
| `log:entry` | Server -> Client | New log entry |
| `set_level` | Client -> Server | Set minimum log level filter |
| `set_filter` | Client -> Server | Set log filters (system, component) |

#### Payload Structures

**log:entry**
```json
{
  "timestamp": "2026-01-31T14:30:00.123Z",
  "level": "info",
  "component": "inbound_pipeline",
  "message": "File processing started",
  "metadata": {
    "file_id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c",
    "system": "CCS",
    "worker_id": 3
  }
}
```

**set_level (Client -> Server)**
```json
{
  "level": "debug"
}
```

**set_filter (Client -> Server)**
```json
{
  "systems": ["CCS", "SPI"],
  "components": ["inbound_pipeline", "sta_client"],
  "file_id": "f8a7b6c5-d4e3-2f1a-0b9c-8d7e6f5a4b3c"
}
```

---

### 4. health:status - Health Updates

Real-time health status monitoring.

#### Events

| Event | Direction | Description |
|-------|-----------|-------------|
| `health:update` | Server -> Client | Health status change |
| `health:degraded` | Server -> Client | System health degraded |
| `health:recovered` | Server -> Client | System health recovered |
| `get_status` | Client -> Server | Request current health status |

#### Payload Structures

**health:update**
```json
{
  "status": "healthy",
  "timestamp": "2026-01-31T14:30:00Z",
  "checks": {
    "database": {
      "status": "healthy",
      "latency_ms": 5
    },
    "sta_connection": {
      "status": "healthy",
      "latency_ms": 120
    },
    "inbound_pipeline": {
      "status": "healthy",
      "workers_active": 4,
      "workers_total": 4
    },
    "outbound_pipeline": {
      "status": "healthy",
      "queue_depth": 5
    }
  }
}
```

**health:degraded**
```json
{
  "status": "degraded",
  "timestamp": "2026-01-31T14:30:00Z",
  "affected_components": ["sta_connection"],
  "message": "STA connection experiencing high latency",
  "checks": {
    "sta_connection": {
      "status": "degraded",
      "latency_ms": 5000,
      "error": "Connection timeout"
    }
  }
}
```

---

### 5. admin:notifications - Admin Alerts

Administrative notifications and system alerts.

#### Events

| Event | Direction | Description |
|-------|-----------|-------------|
| `alert:critical` | Server -> Client | Critical system alert |
| `alert:warning` | Server -> Client | Warning notification |
| `alert:info` | Server -> Client | Informational notification |
| `config:updated` | Server -> Client | Configuration changed |
| `route:toggled` | Server -> Client | Route enabled/disabled |
| `acknowledge` | Client -> Server | Acknowledge an alert |

#### Payload Structures

**alert:critical**
```json
{
  "id": "alert-001",
  "severity": "critical",
  "title": "STA Connection Failed",
  "message": "Unable to connect to STA WebService. Circuit breaker opened.",
  "component": "sta_client",
  "timestamp": "2026-01-31T14:30:00Z",
  "requires_acknowledgement": true,
  "actions": [
    {
      "label": "View Logs",
      "action": "navigate",
      "target": "/admin/logs?component=sta_client"
    },
    {
      "label": "Retry Connection",
      "action": "api_call",
      "endpoint": "/admin/sta/reconnect"
    }
  ]
}
```

**config:updated**
```json
{
  "section": "inbound",
  "changes": {
    "poll_interval": {
      "old": "30s",
      "new": "60s"
    },
    "worker_count": {
      "old": 4,
      "new": 8
    }
  },
  "updated_by": "admin@example.com",
  "timestamp": "2026-01-31T14:30:00Z"
}
```

**route:toggled**
```json
{
  "route_id": "ccs-inbound",
  "system": "CCS",
  "direction": "inbound",
  "enabled": false,
  "toggled_by": "admin@example.com",
  "reason": "Scheduled maintenance",
  "timestamp": "2026-01-31T14:30:00Z"
}
```

---

## JavaScript Client Examples

### Installation

```bash
npm install phoenix
```

### Basic Connection

```javascript
import { Socket } from "phoenix";

// Get token from your authentication endpoint
const token = await fetchSocketToken();

// Create socket connection
const socket = new Socket("ws://localhost:4000/socket/websocket", {
  params: { token: token },
  logger: (kind, msg, data) => {
    console.log(`${kind}: ${msg}`, data);
  }
});

// Connect to the server
socket.connect();

// Handle connection events
socket.onOpen(() => console.log("Connected to server"));
socket.onError((error) => console.error("Socket error:", error));
socket.onClose(() => console.log("Disconnected from server"));
```

### Subscribing to File Events

```javascript
import { Socket } from "phoenix";

class FileMonitor {
  constructor(token) {
    this.socket = new Socket("ws://localhost:4000/socket/websocket", {
      params: { token }
    });
    this.filesChannel = null;
    this.callbacks = {
      onCreated: [],
      onUpdated: [],
      onCompleted: [],
      onFailed: []
    };
  }

  connect() {
    return new Promise((resolve, reject) => {
      this.socket.connect();

      this.filesChannel = this.socket.channel("files:lobby", {});

      this.filesChannel.join()
        .receive("ok", (resp) => {
          console.log("Joined files channel", resp);
          this.setupEventHandlers();
          resolve(this);
        })
        .receive("error", (resp) => {
          console.error("Failed to join files channel", resp);
          reject(resp);
        });
    });
  }

  setupEventHandlers() {
    this.filesChannel.on("file:created", (payload) => {
      this.callbacks.onCreated.forEach(cb => cb(payload));
    });

    this.filesChannel.on("file:updated", (payload) => {
      this.callbacks.onUpdated.forEach(cb => cb(payload));
    });

    this.filesChannel.on("file:completed", (payload) => {
      this.callbacks.onCompleted.forEach(cb => cb(payload));
    });

    this.filesChannel.on("file:failed", (payload) => {
      this.callbacks.onFailed.forEach(cb => cb(payload));
    });
  }

  onFileCreated(callback) {
    this.callbacks.onCreated.push(callback);
    return this;
  }

  onFileUpdated(callback) {
    this.callbacks.onUpdated.push(callback);
    return this;
  }

  onFileCompleted(callback) {
    this.callbacks.onCompleted.push(callback);
    return this;
  }

  onFileFailed(callback) {
    this.callbacks.onFailed.push(callback);
    return this;
  }

  subscribeToFile(fileId) {
    this.filesChannel.push("subscribe_file", { file_id: fileId })
      .receive("ok", () => console.log(`Subscribed to file ${fileId}`))
      .receive("error", (err) => console.error("Subscribe failed:", err));
  }

  disconnect() {
    this.filesChannel?.leave();
    this.socket.disconnect();
  }
}

// Usage
const monitor = new FileMonitor(token);

await monitor.connect();

monitor
  .onFileCreated((file) => {
    console.log("New file:", file.filename);
    updateFileList(file);
  })
  .onFileUpdated((file) => {
    console.log(`File ${file.id} status: ${file.status}`);
    updateFileStatus(file);
  })
  .onFileCompleted((file) => {
    console.log(`File ${file.id} completed in ${file.processing_time_ms}ms`);
    showSuccessNotification(file);
  })
  .onFileFailed((file) => {
    console.error(`File ${file.id} failed: ${file.error_message}`);
    showErrorNotification(file);
  });
```

### Real-time Metrics Dashboard

```javascript
import { Socket } from "phoenix";

class MetricsDashboard {
  constructor(token) {
    this.socket = new Socket("ws://localhost:4000/socket/websocket", {
      params: { token }
    });
    this.metricsChannel = null;
  }

  async connect() {
    this.socket.connect();

    this.metricsChannel = this.socket.channel("metrics:lobby", {});

    return new Promise((resolve, reject) => {
      this.metricsChannel.join()
        .receive("ok", () => resolve(this))
        .receive("error", reject);
    });
  }

  onMetricsUpdate(callback) {
    this.metricsChannel.on("metrics:update", callback);
    return this;
  }

  onThresholdAlert(callback) {
    this.metricsChannel.on("metrics:threshold_alert", callback);
    return this;
  }

  requestCurrentMetrics() {
    return new Promise((resolve, reject) => {
      this.metricsChannel.push("get_current", {})
        .receive("ok", resolve)
        .receive("error", reject);
    });
  }
}

// Usage
const dashboard = new MetricsDashboard(token);
await dashboard.connect();

dashboard
  .onMetricsUpdate((metrics) => {
    updateInboundChart(metrics.inbound);
    updateOutboundChart(metrics.outbound);
    updateSystemStats(metrics.system);
  })
  .onThresholdAlert((alert) => {
    showAlertBanner(alert);
  });

// Get initial metrics
const currentMetrics = await dashboard.requestCurrentMetrics();
initializeCharts(currentMetrics);
```

### Health Status Monitor

```javascript
import { Socket } from "phoenix";

class HealthMonitor {
  constructor(token) {
    this.socket = new Socket("ws://localhost:4000/socket/websocket", {
      params: { token }
    });
    this.healthChannel = null;
  }

  async connect() {
    this.socket.connect();

    this.healthChannel = this.socket.channel("health:status", {});

    return new Promise((resolve, reject) => {
      this.healthChannel.join()
        .receive("ok", () => resolve(this))
        .receive("error", reject);
    });
  }

  onHealthUpdate(callback) {
    this.healthChannel.on("health:update", callback);
    return this;
  }

  onDegraded(callback) {
    this.healthChannel.on("health:degraded", callback);
    return this;
  }

  onRecovered(callback) {
    this.healthChannel.on("health:recovered", callback);
    return this;
  }
}

// Usage
const health = new HealthMonitor(token);
await health.connect();

health
  .onHealthUpdate((status) => {
    updateHealthIndicator(status.status);
    updateComponentStatus(status.checks);
  })
  .onDegraded((status) => {
    showDegradedBanner(status.affected_components, status.message);
    playAlertSound();
  })
  .onRecovered((status) => {
    hideDegradedBanner();
    showRecoveryNotification();
  });
```

### Admin Notifications

```javascript
import { Socket } from "phoenix";

class AdminNotifications {
  constructor(token) {
    this.socket = new Socket("ws://localhost:4000/socket/websocket", {
      params: { token }
    });
    this.adminChannel = null;
  }

  async connect() {
    this.socket.connect();

    this.adminChannel = this.socket.channel("admin:notifications", {});

    return new Promise((resolve, reject) => {
      this.adminChannel.join()
        .receive("ok", () => resolve(this))
        .receive("error", reject);
    });
  }

  onCriticalAlert(callback) {
    this.adminChannel.on("alert:critical", callback);
    return this;
  }

  onWarning(callback) {
    this.adminChannel.on("alert:warning", callback);
    return this;
  }

  onInfo(callback) {
    this.adminChannel.on("alert:info", callback);
    return this;
  }

  onConfigUpdated(callback) {
    this.adminChannel.on("config:updated", callback);
    return this;
  }

  onRouteToggled(callback) {
    this.adminChannel.on("route:toggled", callback);
    return this;
  }

  acknowledgeAlert(alertId) {
    return new Promise((resolve, reject) => {
      this.adminChannel.push("acknowledge", { alert_id: alertId })
        .receive("ok", resolve)
        .receive("error", reject);
    });
  }
}

// Usage
const notifications = new AdminNotifications(token);
await notifications.connect();

notifications
  .onCriticalAlert((alert) => {
    showCriticalModal(alert);
    sendSlackNotification(alert);
  })
  .onWarning((alert) => {
    addToNotificationCenter(alert);
  })
  .onConfigUpdated((change) => {
    console.log("Config updated:", change.section, change.changes);
    refreshConfigDisplay();
  })
  .onRouteToggled((route) => {
    updateRouteStatus(route.route_id, route.enabled);
  });
```

### Complete Integration Example

```javascript
import { Socket } from "phoenix";

class STAConnectorClient {
  constructor(endpoint, token) {
    this.socket = new Socket(endpoint, { params: { token } });
    this.channels = {};
  }

  async connect() {
    this.socket.connect();

    // Join all channels
    const channelConfigs = [
      { name: "files:lobby", key: "files" },
      { name: "metrics:lobby", key: "metrics" },
      { name: "logs:lobby", key: "logs" },
      { name: "health:status", key: "health" },
      { name: "admin:notifications", key: "admin" }
    ];

    for (const config of channelConfigs) {
      this.channels[config.key] = await this.joinChannel(config.name);
    }

    return this;
  }

  joinChannel(name) {
    return new Promise((resolve, reject) => {
      const channel = this.socket.channel(name, {});
      channel.join()
        .receive("ok", () => resolve(channel))
        .receive("error", reject);
    });
  }

  // File events
  onFile(event, callback) {
    this.channels.files.on(`file:${event}`, callback);
    return this;
  }

  // Metrics events
  onMetrics(event, callback) {
    this.channels.metrics.on(`metrics:${event}`, callback);
    return this;
  }

  // Log events
  onLog(callback) {
    this.channels.logs.on("log:entry", callback);
    return this;
  }

  setLogLevel(level) {
    this.channels.logs.push("set_level", { level });
    return this;
  }

  setLogFilter(filter) {
    this.channels.logs.push("set_filter", filter);
    return this;
  }

  // Health events
  onHealth(event, callback) {
    this.channels.health.on(`health:${event}`, callback);
    return this;
  }

  // Admin events
  onAlert(severity, callback) {
    this.channels.admin.on(`alert:${severity}`, callback);
    return this;
  }

  onConfigChange(callback) {
    this.channels.admin.on("config:updated", callback);
    return this;
  }

  disconnect() {
    Object.values(this.channels).forEach(ch => ch.leave());
    this.socket.disconnect();
  }
}

// Usage
const client = new STAConnectorClient("ws://localhost:4000/socket/websocket", token);
await client.connect();

client
  .onFile("created", (f) => console.log("New file:", f.filename))
  .onFile("completed", (f) => console.log("Completed:", f.id))
  .onFile("failed", (f) => console.error("Failed:", f.id, f.error_message))
  .onMetrics("update", (m) => updateDashboard(m))
  .onHealth("degraded", (h) => showAlert(h))
  .onAlert("critical", (a) => sendUrgentNotification(a))
  .onLog((entry) => appendToLogViewer(entry));

// Filter logs to only CCS system, debug level
client.setLogLevel("debug");
client.setLogFilter({ systems: ["CCS"] });
```

---

## Server-side Elixir Examples

### Channel Implementations

#### Files Channel

```elixir
# lib/sta_connector_web/channels/files_channel.ex
defmodule STAConnectorWeb.FilesChannel do
  use STAConnectorWeb, :channel

  @impl true
  def join("files:lobby", _payload, socket) do
    {:ok, socket}
  end

  @impl true
  def handle_in("subscribe_file", %{"file_id" => file_id}, socket) do
    # Subscribe to file-specific updates
    :ok = Phoenix.PubSub.subscribe(STAConnector.PubSub, "file:#{file_id}")
    {:reply, :ok, assign(socket, :subscribed_files, [file_id | socket.assigns[:subscribed_files] || []])}
  end

  @impl true
  def handle_in("unsubscribe_file", %{"file_id" => file_id}, socket) do
    :ok = Phoenix.PubSub.unsubscribe(STAConnector.PubSub, "file:#{file_id}")
    subscribed = List.delete(socket.assigns[:subscribed_files] || [], file_id)
    {:reply, :ok, assign(socket, :subscribed_files, subscribed)}
  end
end
```

#### Metrics Channel

```elixir
# lib/sta_connector_web/channels/metrics_channel.ex
defmodule STAConnectorWeb.MetricsChannel do
  use STAConnectorWeb, :channel

  alias STAConnector.Metrics

  @impl true
  def join("metrics:lobby", _payload, socket) do
    # Send initial metrics on join
    send(self(), :send_initial_metrics)
    {:ok, socket}
  end

  @impl true
  def handle_info(:send_initial_metrics, socket) do
    metrics = Metrics.get_current()
    push(socket, "metrics:update", metrics)
    {:noreply, socket}
  end

  @impl true
  def handle_in("get_current", _payload, socket) do
    metrics = Metrics.get_current()
    {:reply, {:ok, metrics}, socket}
  end
end
```

#### Logs Channel

```elixir
# lib/sta_connector_web/channels/logs_channel.ex
defmodule STAConnectorWeb.LogsChannel do
  use STAConnectorWeb, :channel

  @impl true
  def join("logs:lobby", _payload, socket) do
    {:ok, assign(socket, :log_level, :info, :log_filter, %{})}
  end

  @impl true
  def handle_in("set_level", %{"level" => level}, socket) do
    level_atom = String.to_existing_atom(level)
    {:reply, :ok, assign(socket, :log_level, level_atom)}
  end

  @impl true
  def handle_in("set_filter", filter, socket) do
    {:reply, :ok, assign(socket, :log_filter, filter)}
  end
end
```

#### Health Channel

```elixir
# lib/sta_connector_web/channels/health_channel.ex
defmodule STAConnectorWeb.HealthChannel do
  use STAConnectorWeb, :channel

  alias STAConnector.Health

  @impl true
  def join("health:status", _payload, socket) do
    # Send initial health status
    send(self(), :send_initial_health)
    {:ok, socket}
  end

  @impl true
  def handle_info(:send_initial_health, socket) do
    health = Health.get_status()
    push(socket, "health:update", health)
    {:noreply, socket}
  end

  @impl true
  def handle_in("get_status", _payload, socket) do
    health = Health.get_status()
    {:reply, {:ok, health}, socket}
  end
end
```

#### Admin Channel

```elixir
# lib/sta_connector_web/channels/admin_channel.ex
defmodule STAConnectorWeb.AdminChannel do
  use STAConnectorWeb, :channel

  alias STAConnector.Alerts

  @impl true
  def join("admin:notifications", _payload, socket) do
    # Verify admin permissions
    if socket.assigns[:user_id] && has_admin_role?(socket.assigns.user_id) do
      {:ok, socket}
    else
      {:error, %{reason: "unauthorized"}}
    end
  end

  @impl true
  def handle_in("acknowledge", %{"alert_id" => alert_id}, socket) do
    case Alerts.acknowledge(alert_id, socket.assigns.user_id) do
      :ok -> {:reply, :ok, socket}
      {:error, reason} -> {:reply, {:error, %{reason: reason}}, socket}
    end
  end

  defp has_admin_role?(user_id) do
    # Check user permissions
    STAConnector.Users.has_role?(user_id, :admin)
  end
end
```

### Broadcasting Events

#### File Events Broadcasting

```elixir
# lib/sta_connector/file_processor.ex
defmodule STAConnector.FileProcessor do
  alias STAConnectorWeb.Endpoint

  def process_file(file) do
    # Broadcast file created
    broadcast_file_event("file:created", %{
      id: file.id,
      filename: file.filename,
      system: file.system,
      file_type: file.file_type,
      size: file.size,
      source_ispb: file.source_ispb,
      destination_ispb: file.destination_ispb,
      direction: file.direction,
      status: "pending",
      created_at: DateTime.utc_now()
    })

    # Process file...
    result = do_process(file)

    case result do
      {:ok, processed_file} ->
        broadcast_file_event("file:completed", %{
          id: file.id,
          status: "completed",
          processed_at: DateTime.utc_now(),
          processing_time_ms: processed_file.processing_time_ms,
          records_processed: processed_file.records_processed,
          checksum: processed_file.checksum
        })

      {:error, reason} ->
        broadcast_file_event("file:failed", %{
          id: file.id,
          status: "failed",
          error_code: reason.code,
          error_message: reason.message,
          failed_at: DateTime.utc_now(),
          retry_count: file.retry_count,
          max_retries: 5
        })
    end
  end

  def update_file_progress(file_id, progress) do
    broadcast_file_event("file:updated", %{
      id: file_id,
      status: "processing",
      progress: progress,
      updated_at: DateTime.utc_now()
    })
  end

  defp broadcast_file_event(event, payload) do
    Endpoint.broadcast("files:lobby", event, payload)
  end
end
```

#### Metrics Broadcasting

```elixir
# lib/sta_connector/metrics_broadcaster.ex
defmodule STAConnector.MetricsBroadcaster do
  use GenServer

  alias STAConnectorWeb.Endpoint
  alias STAConnector.Metrics

  @broadcast_interval 5_000  # 5 seconds

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do
    schedule_broadcast()
    {:ok, %{}}
  end

  @impl true
  def handle_info(:broadcast, state) do
    metrics = Metrics.get_current()
    Endpoint.broadcast("metrics:lobby", "metrics:update", metrics)

    # Check thresholds
    check_thresholds(metrics)

    schedule_broadcast()
    {:noreply, state}
  end

  defp schedule_broadcast do
    Process.send_after(self(), :broadcast, @broadcast_interval)
  end

  defp check_thresholds(metrics) do
    thresholds = [
      {:inbound_pending, metrics.inbound.files_pending, 100, "warning"},
      {:rate_limit, metrics.sta_client.rate_limit_remaining, 10, "warning"},
      {:failure_rate, calculate_failure_rate(metrics), 0.05, "critical"}
    ]

    for {metric, value, threshold, severity} <- thresholds,
        exceeds_threshold?(metric, value, threshold) do
      Endpoint.broadcast("metrics:lobby", "metrics:threshold_alert", %{
        metric: to_string(metric),
        current_value: value,
        threshold: threshold,
        severity: severity,
        message: threshold_message(metric),
        triggered_at: DateTime.utc_now()
      })
    end
  end

  defp exceeds_threshold?(:rate_limit, value, threshold), do: value < threshold
  defp exceeds_threshold?(_metric, value, threshold), do: value > threshold

  defp calculate_failure_rate(%{inbound: %{files_processed: 0}}), do: 0.0
  defp calculate_failure_rate(%{inbound: %{files_processed: total, files_failed: failed}}) do
    failed / total
  end

  defp threshold_message(:inbound_pending), do: "High pending file count detected"
  defp threshold_message(:rate_limit), do: "STA rate limit nearly exhausted"
  defp threshold_message(:failure_rate), do: "High failure rate detected (>5%)"
end
```

#### Health Status Broadcasting

```elixir
# lib/sta_connector/health_broadcaster.ex
defmodule STAConnector.HealthBroadcaster do
  use GenServer

  alias STAConnectorWeb.Endpoint
  alias STAConnector.Health

  @check_interval 10_000  # 10 seconds

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do
    schedule_check()
    {:ok, %{last_status: "healthy"}}
  end

  @impl true
  def handle_info(:check_health, state) do
    health = Health.get_status()
    new_state = handle_status_change(health, state)

    Endpoint.broadcast("health:status", "health:update", health)

    schedule_check()
    {:noreply, new_state}
  end

  defp handle_status_change(%{status: "degraded"} = health, %{last_status: "healthy"}) do
    Endpoint.broadcast("health:status", "health:degraded", %{
      status: "degraded",
      timestamp: DateTime.utc_now(),
      affected_components: get_unhealthy_components(health.checks),
      message: "System health degraded",
      checks: health.checks
    })
    %{last_status: "degraded"}
  end

  defp handle_status_change(%{status: "healthy"} = health, %{last_status: status}) when status != "healthy" do
    Endpoint.broadcast("health:status", "health:recovered", %{
      status: "healthy",
      timestamp: DateTime.utc_now(),
      checks: health.checks
    })
    %{last_status: "healthy"}
  end

  defp handle_status_change(_health, state), do: state

  defp get_unhealthy_components(checks) do
    checks
    |> Enum.filter(fn {_name, check} -> check.status != "healthy" end)
    |> Enum.map(fn {name, _check} -> to_string(name) end)
  end

  defp schedule_check do
    Process.send_after(self(), :check_health, @check_interval)
  end
end
```

#### Admin Alert Broadcasting

```elixir
# lib/sta_connector/alerts.ex
defmodule STAConnector.Alerts do
  alias STAConnectorWeb.Endpoint

  def broadcast_critical(title, message, component, opts \\ []) do
    alert = build_alert(:critical, title, message, component, opts)
    Endpoint.broadcast("admin:notifications", "alert:critical", alert)
    store_alert(alert)
  end

  def broadcast_warning(title, message, component, opts \\ []) do
    alert = build_alert(:warning, title, message, component, opts)
    Endpoint.broadcast("admin:notifications", "alert:warning", alert)
    store_alert(alert)
  end

  def broadcast_info(title, message, component, opts \\ []) do
    alert = build_alert(:info, title, message, component, opts)
    Endpoint.broadcast("admin:notifications", "alert:info", alert)
  end

  def broadcast_config_change(section, changes, updated_by) do
    Endpoint.broadcast("admin:notifications", "config:updated", %{
      section: section,
      changes: changes,
      updated_by: updated_by,
      timestamp: DateTime.utc_now()
    })
  end

  def broadcast_route_toggle(route_id, system, direction, enabled, toggled_by, reason \\ nil) do
    Endpoint.broadcast("admin:notifications", "route:toggled", %{
      route_id: route_id,
      system: system,
      direction: direction,
      enabled: enabled,
      toggled_by: toggled_by,
      reason: reason,
      timestamp: DateTime.utc_now()
    })
  end

  def acknowledge(alert_id, user_id) do
    # Mark alert as acknowledged in database
    case update_alert_status(alert_id, :acknowledged, user_id) do
      {:ok, _} -> :ok
      error -> error
    end
  end

  defp build_alert(severity, title, message, component, opts) do
    %{
      id: generate_alert_id(),
      severity: to_string(severity),
      title: title,
      message: message,
      component: component,
      timestamp: DateTime.utc_now(),
      requires_acknowledgement: Keyword.get(opts, :requires_ack, severity == :critical),
      actions: Keyword.get(opts, :actions, [])
    }
  end

  defp generate_alert_id do
    "alert-#{:crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower)}"
  end

  defp store_alert(alert) do
    # Store in database for persistence
    STAConnector.Repo.insert(%STAConnector.Alert{
      id: alert.id,
      severity: alert.severity,
      title: alert.title,
      message: alert.message,
      component: alert.component,
      requires_acknowledgement: alert.requires_acknowledgement
    })
  end

  defp update_alert_status(alert_id, status, user_id) do
    # Update alert status in database
    STAConnector.Alert
    |> STAConnector.Repo.get(alert_id)
    |> STAConnector.Alert.changeset(%{status: status, acknowledged_by: user_id, acknowledged_at: DateTime.utc_now()})
    |> STAConnector.Repo.update()
  end
end
```

### Log Streaming

```elixir
# lib/sta_connector/log_broadcaster.ex
defmodule STAConnector.LogBroadcaster do
  @moduledoc """
  Broadcasts log entries to connected WebSocket clients.
  Attach this as a Logger backend or call directly.
  """

  alias STAConnectorWeb.Endpoint

  def broadcast_log(level, message, metadata \\ %{}) do
    entry = %{
      timestamp: DateTime.utc_now() |> DateTime.to_iso8601(),
      level: to_string(level),
      component: Map.get(metadata, :component, "unknown"),
      message: message,
      metadata: Map.drop(metadata, [:component])
    }

    Endpoint.broadcast("logs:lobby", "log:entry", entry)
  end

  # Logger backend implementation
  def handle_event({level, _gl, {Logger, message, timestamp, metadata}}, state) do
    broadcast_log(level, IO.iodata_to_binary(message), Map.new(metadata))
    {:ok, state}
  end
end
```

---

## Error Handling

### Client-side Error Handling

```javascript
const socket = new Socket("ws://localhost:4000/socket/websocket", {
  params: { token }
});

// Handle socket-level errors
socket.onError((error) => {
  console.error("Socket error:", error);
  showReconnectingBanner();
});

socket.onClose((event) => {
  console.log("Socket closed:", event);
  if (event.code === 1006) {
    // Abnormal closure - attempt reconnect
    setTimeout(() => socket.connect(), 5000);
  }
});

// Handle channel-level errors
const channel = socket.channel("files:lobby", {});

channel.join()
  .receive("ok", () => console.log("Joined successfully"))
  .receive("error", (resp) => {
    if (resp.reason === "unauthorized") {
      redirectToLogin();
    } else {
      showErrorNotification(`Failed to join channel: ${resp.reason}`);
    }
  })
  .receive("timeout", () => {
    showErrorNotification("Connection timeout - retrying...");
  });

// Handle push errors
channel.push("subscribe_file", { file_id: "123" })
  .receive("ok", () => console.log("Subscribed"))
  .receive("error", (resp) => console.error("Subscribe failed:", resp))
  .receive("timeout", () => console.error("Request timed out"));
```

### Reconnection Strategy

```javascript
class ReconnectingSocket {
  constructor(endpoint, token, options = {}) {
    this.endpoint = endpoint;
    this.token = token;
    this.maxRetries = options.maxRetries || 10;
    this.retryInterval = options.retryInterval || 5000;
    this.retryCount = 0;
    this.socket = null;
    this.channels = new Map();
  }

  connect() {
    this.socket = new Socket(this.endpoint, {
      params: { token: this.token }
    });

    this.socket.onClose(() => this.handleDisconnect());
    this.socket.onError(() => this.handleDisconnect());

    this.socket.connect();
    this.retryCount = 0;

    // Rejoin all channels
    this.channels.forEach((callbacks, channelName) => {
      this.joinChannel(channelName, callbacks);
    });
  }

  handleDisconnect() {
    if (this.retryCount < this.maxRetries) {
      this.retryCount++;
      console.log(`Reconnecting (attempt ${this.retryCount}/${this.maxRetries})...`);
      setTimeout(() => this.connect(), this.retryInterval * this.retryCount);
    } else {
      console.error("Max reconnection attempts reached");
      this.onMaxRetriesExceeded?.();
    }
  }

  joinChannel(name, callbacks = {}) {
    this.channels.set(name, callbacks);

    const channel = this.socket.channel(name, {});

    Object.entries(callbacks).forEach(([event, callback]) => {
      channel.on(event, callback);
    });

    channel.join()
      .receive("ok", () => console.log(`Joined ${name}`))
      .receive("error", (resp) => console.error(`Failed to join ${name}:`, resp));

    return channel;
  }

  onMaxRetriesExceeded(callback) {
    this.onMaxRetriesExceeded = callback;
  }
}
```

---

## Best Practices

### 1. Token Refresh

```javascript
class TokenManager {
  constructor(refreshEndpoint) {
    this.refreshEndpoint = refreshEndpoint;
    this.token = null;
    this.refreshTimer = null;
  }

  async getToken() {
    if (!this.token || this.isExpiringSoon()) {
      await this.refresh();
    }
    return this.token;
  }

  async refresh() {
    const response = await fetch(this.refreshEndpoint, {
      method: "POST",
      credentials: "include"
    });
    const data = await response.json();
    this.token = data.token;
    this.scheduleRefresh(data.expires_in);
  }

  isExpiringSoon() {
    // Check if token expires in less than 5 minutes
    return this.expiresAt && (this.expiresAt - Date.now()) < 300000;
  }

  scheduleRefresh(expiresIn) {
    // Refresh 5 minutes before expiration
    const refreshIn = (expiresIn - 300) * 1000;
    clearTimeout(this.refreshTimer);
    this.refreshTimer = setTimeout(() => this.refresh(), refreshIn);
  }
}
```

### 2. Presence Tracking

```elixir
# Track which users are viewing which files
defmodule STAConnectorWeb.Presence do
  use Phoenix.Presence,
    otp_app: :sta_connector,
    pubsub_server: STAConnector.PubSub
end

# In FilesChannel
def handle_in("start_viewing", %{"file_id" => file_id}, socket) do
  {:ok, _} = Presence.track(socket, file_id, %{
    user_id: socket.assigns.user_id,
    joined_at: DateTime.utc_now()
  })
  {:reply, :ok, socket}
end
```

### 3. Rate Limiting

```elixir
# Rate limit channel messages
defmodule STAConnectorWeb.RateLimiter do
  use GenServer

  @max_messages_per_second 10

  def check_rate(user_id) do
    case :ets.update_counter(:rate_limits, user_id, {2, 1}, {user_id, 0, now()}) do
      count when count > @max_messages_per_second ->
        {:error, :rate_limited}
      _ ->
        :ok
    end
  end
end
```

---

## Related Documentation

- [Admin API Reference](/docs/api/admin.md) - REST API documentation
- [Outbound API Reference](/docs/api/outbound.md) - File submission API
- [Authentication Guide](/docs/guide/authentication.md) - Token and auth setup
- [Deployment Guide](/docs/deployment/README.md) - Production configuration