# Inbound Pipeline

The inbound pipeline automatically polls BCB STA for available files, downloads them, parses them, and delivers them to your configured Monetarie STA endpoints.

## Architecture Overview

```
                    ┌─────────────────┐
                    │   BCB STA       │
                    │   WebService    │
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │  STA Client     │
                    │ (SOAP/HTTPS)    │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
     ┌────────▼────────┐     │     ┌────────▼────────┐
     │   Poller        │     │     │  Rate Limiter   │
     │ (GenServer)     │     │     │ (10 concurrent) │
     └────────┬────────┘     │     └─────────────────┘
              │              │
     ┌────────▼────────┐     │
     │ Task.Supervisor │◀────┘
     │ (Download/Parse)│
     └────────┬────────┘
              │
     ┌────────▼────────┐
     │   Router        │
     │ (Finch HTTP)    │
     └────────┬────────┘
              │
     ┌────────▼────────┐
     │   Monetarie STA    │
     │   Endpoints     │
     └─────────────────┘
```

## How It Works

### 1. Polling for Files

The `StaConnector.Files.Poller` GenServer runs at configured intervals and queries BCB for available files:

```elixir
defmodule StaConnector.Files.Poller do
  use GenServer

  @impl true
  def init(opts) do
    interval = Keyword.get(opts, :interval, :timer.seconds(30))
    schedule_poll(interval)
    {:ok, %{interval: interval, systems: opts[:systems] || []}}
  end

  @impl true
  def handle_info(:poll, state) do
    # Query BCB for available files
    case StaConnector.Sta.Client.query_available_files(state.systems) do
      {:ok, files} ->
        Enum.each(files, &process_file/1)

      {:error, reason} ->
        Logger.error("Failed to poll STA: #{inspect(reason)}")
    end

    schedule_poll(state.interval)
    {:noreply, state}
  end

  defp schedule_poll(interval) do
    Process.send_after(self(), :poll, interval)
  end
end
```

### 2. OTP Supervision Tree

The inbound pipeline is supervised for fault tolerance:

```elixir
defmodule StaConnector.Application do
  use Application

  def start(_type, _args) do
    children = [
      # Database repo
      StaConnector.Repo,

      # HTTP client pool
      {Finch, name: StaConnector.Finch},

      # Task supervisor for file processing
      {Task.Supervisor, name: StaConnector.TaskSupervisor},

      # Phoenix PubSub for real-time updates
      {Phoenix.PubSub, name: StaConnector.PubSub},

      # File poller GenServer
      {StaConnector.Files.Poller, interval: poll_interval(), systems: systems()},

      # Phoenix endpoint
      StaConnectorWeb.Endpoint
    ]

    opts = [strategy: :one_for_one, name: StaConnector.Supervisor]
    Supervisor.start_link(children, opts)
  end

  defp poll_interval do
    Application.get_env(:sta_connector, :poller)[:interval_seconds]
    |> Kernel.*(1000)
  end

  defp systems do
    Application.get_env(:sta_connector, :poller)[:systems] || []
  end
end
```

### 3. Processing Files with Task.Supervisor

For each discovered file, tasks are spawned under supervision:

```elixir
defmodule StaConnector.Files.Processor do
  alias StaConnector.{Repo, Files.File, Sta.Client, Router}

  def process_file(sta_file) do
    Task.Supervisor.start_child(StaConnector.TaskSupervisor, fn ->
      with {:ok, file} <- create_or_get_file(sta_file),
           {:ok, file} <- download_content(file),
           {:ok, file} <- parse_content(file),
           {:ok, file} <- deliver_to_route(file),
           {:ok, _} <- confirm_receipt(file) do
        complete_file(file)
      else
        {:error, reason} ->
          fail_file(sta_file, reason)
      end
    end)
  end

  defp download_content(file) do
    file
    |> File.changeset(%{status: :downloading})
    |> Repo.update!()
    |> broadcast_update()

    case Client.download_file(file.protocol_number) do
      {:ok, content} ->
        {:ok, Repo.update!(File.changeset(file, %{content: content}))}

      error ->
        error
    end
  end

  defp broadcast_update(file) do
    Phoenix.PubSub.broadcast(
      StaConnector.PubSub,
      "files:lobby",
      {:file_updated, file}
    )
    file
  end
end
```

### 4. Delivery to Monetarie STA with Finch

Files are delivered via HTTP using the Finch client:

```elixir
defmodule StaConnector.Router do
  def deliver(file) do
    route = get_route(file.system_id)

    request =
      Finch.build(
        :post,
        route.url,
        build_headers(route),
        Jason.encode!(build_payload(file))
      )

    case Finch.request(request, StaConnector.Finch, receive_timeout: route.timeout_ms) do
      {:ok, %Finch.Response{status: status}} when status in 200..299 ->
        {:ok, file}

      {:ok, %Finch.Response{status: status, body: body}} ->
        {:error, {:http_error, status, body}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp build_headers(route) do
    [
      {"content-type", "application/json"},
      {"authorization", route.auth_header}
      | route.custom_headers
    ]
  end
end
```

### 5. Receipt Confirmation

After successful delivery, the connector confirms with BCB:

```elixir
defmodule StaConnector.Sta.Client do
  def confirm_receipt(protocol_number) do
    # Confirm receipt removes file from BCB queue
    case call_sta_service(:confirmar_recebimento, %{protocolo: protocol_number}) do
      {:ok, _response} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end
end
```

## Configuration

### Basic Setup

Enable the inbound poller in `config/config.exs`:

```elixir
config :sta_connector, :poller,
  enabled: true,
  interval_seconds: 30,
  batch_size: 10
```

### System Selection

Specify which BCB systems to poll:

```elixir
config :sta_connector, :poller,
  systems: ["CCS", "SPI", "DICT"]
```

Leave empty or omit to poll all supported systems.

### Route Configuration

Define delivery routes for each system in `config/config.exs`:

```elixir
config :sta_connector, :routes,
  SPI: %{
    url: "https://staapi-planner.monetarie.com.br/api/v1/spi",
    method: :post,
    headers: [
      {"authorization", "Bearer #{System.get_env("MONETARIE_STA_TOKEN")}"},
      {"content-type", "application/json"}
    ],
    timeout_ms: 30_000,
    retry_attempts: 3,
    enabled: true
  }
```

### Environment-Specific Configuration

Use `config/runtime.exs` for runtime configuration:

```elixir
import Config

if config_env() == :prod do
  config :sta_connector, :sta,
    endpoint: System.fetch_env!("STA_ENDPOINT"),
    username: System.fetch_env!("STA_USERNAME"),
    password: System.fetch_env!("STA_PASSWORD")

  config :sta_connector, :routes,
    SPI: %{
      url: System.fetch_env!("MONETARIE_STA_SPI_URL"),
      headers: [
        {"authorization", "Bearer #{System.fetch_env!("MONETARIE_STA_TOKEN")}"}
      ]
    }
end
```

## File Processing States

Files move through the following states:

```
┌─────────┐    ┌────────────┐    ┌────────────┐    ┌───────────┐
│ pending │───▶│ downloading│───▶│ processing │───▶│ completed │
└─────────┘    └─────┬──────┘    └─────┬──────┘    └───────────┘
                     │                  │
                     │                  │
                     ▼                  ▼
               ┌──────────┐      ┌──────────┐
               │  failed  │◀─────│  failed  │
               └────┬─────┘      └──────────┘
                    │
                    ▼
               ┌──────────┐
               │ retrying │
               └──────────┘
```

| State | Description |
|-------|-------------|
| `pending` | File discovered, queued for processing |
| `downloading` | Downloading content from BCB |
| `processing` | Parsing and routing |
| `completed` | Successfully delivered and confirmed |
| `failed` | Processing failed |
| `retrying` | Scheduled for retry |

## Ecto Schema for File State

```elixir
defmodule StaConnector.Files.File do
  use Ecto.Schema
  import Ecto.Changeset

  @statuses [:pending, :downloading, :processing, :completed, :failed, :retrying]

  schema "files" do
    field :protocol_number, :string
    field :system_id, :string
    field :file_name, :string
    field :file_type, :string
    field :content, :binary
    field :status, Ecto.Enum, values: @statuses, default: :pending
    field :error_message, :string
    field :retry_count, :integer, default: 0
    field :checksum, :string
    field :file_size, :integer

    timestamps()
  end

  def changeset(file, attrs) do
    file
    |> cast(attrs, [:protocol_number, :system_id, :file_name, :file_type,
                    :content, :status, :error_message, :retry_count,
                    :checksum, :file_size])
    |> validate_required([:protocol_number, :system_id])
    |> validate_inclusion(:status, @statuses)
    |> unique_constraint(:protocol_number)
  end
end
```

## Real-Time Updates via Phoenix Channels

The admin portal receives real-time file updates via Phoenix Channels:

```elixir
defmodule StaConnectorWeb.FilesChannel do
  use StaConnectorWeb, :channel

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

  @impl true
  def handle_info({:file_updated, file}, socket) do
    push(socket, "file_updated", %{
      id: file.id,
      protocol_number: file.protocol_number,
      system_id: file.system_id,
      status: file.status,
      updated_at: file.updated_at
    })

    {:noreply, socket}
  end
end
```

Subscribe to updates from the processor:

```elixir
# In the processor, broadcast updates
defp broadcast_update(file) do
  Phoenix.PubSub.broadcast(
    StaConnector.PubSub,
    "files:lobby",
    {:file_updated, file}
  )
  file
end
```

## Delivery Options

### HTTP Webhook

Deliver files via HTTP POST:

```elixir
config :sta_connector, :routes,
  SPI: %{
    url: "https://your-api.com/webhooks/sta/spi",
    method: :post,
    headers: [
      {"authorization", "Bearer #{System.get_env("TOKEN")}"},
      {"content-type", "application/json"},
      {"x-custom-header", "value"}
    ],
    timeout_ms: 30_000,
    retry_attempts: 3,
    enabled: true
  }
```

### Webhook Payload

The connector delivers files with the following JSON structure:

```json
{
  "protocol_number": "2026013100001234",
  "system_id": "SPI",
  "file_name": "SPI_20260131_001.xml",
  "file_type": "PACS.008",
  "content": "... base64 or parsed content ...",
  "metadata": {
    "received_at": "2026-01-31T10:30:00Z",
    "file_size": 12345,
    "checksum": "sha256:abc123..."
  }
}
```

## Retry Handling

Configure retry behavior per route:

```elixir
config :sta_connector, :routes,
  SPI: %{
    retry_attempts: 5,
    timeout_ms: 30_000
  }
```

| Setting | Default | Description |
|---------|---------|-------------|
| `retry_attempts` | `3` | Maximum retry attempts |
| `timeout_ms` | `30_000` | Request timeout in milliseconds |

Retries use exponential backoff: 1s, 2s, 4s, 8s, 16s...

### Retry Implementation with GenServer

```elixir
defmodule StaConnector.Files.RetryWorker do
  use GenServer

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

  @impl true
  def handle_info(:check_retries, state) do
    files_to_retry()
    |> Enum.each(&schedule_retry/1)

    schedule_check()
    {:noreply, state}
  end

  defp files_to_retry do
    from(f in File,
      where: f.status == :failed,
      where: f.retry_count < ^max_retries(),
      where: f.updated_at < ago(^backoff_seconds(f.retry_count), "second")
    )
    |> Repo.all()
  end

  defp schedule_retry(file) do
    file
    |> File.changeset(%{status: :retrying, retry_count: file.retry_count + 1})
    |> Repo.update!()

    StaConnector.Files.Processor.process_file(file)
  end

  defp backoff_seconds(retry_count) do
    :math.pow(2, retry_count) |> round()
  end

  defp schedule_check do
    Process.send_after(self(), :check_retries, :timer.seconds(60))
  end
end
```

## Manual Operations

### Retry Failed Files

Retry a specific failed file via Admin API:

```bash
curl -u admin:password -X POST \
  http://localhost:4000/api/admin/retry/{file_id}
```

Or using the LiveView admin interface, which provides real-time feedback.

### Trigger Immediate Poll

Trigger a poll cycle immediately:

```bash
curl -u admin:password -X POST \
  http://localhost:4000/api/admin/poller/trigger
```

Or send a message directly to the GenServer:

```elixir
GenServer.cast(StaConnector.Files.Poller, :poll_now)
```

### List Pending Files

View files pending processing:

```bash
curl -u admin:password \
  "http://localhost:4000/api/admin/files?status=pending"
```

Or query directly with Ecto:

```elixir
from(f in File, where: f.status == :pending)
|> Repo.all()
```

## Monitoring

### Health Check

Check poller status:

```bash
curl http://localhost:4000/api/health
```

```json
{
  "status": "healthy",
  "sta_connection": "ok",
  "database": "ok",
  "details": {
    "pending_files": 5,
    "last_sta_check": "2026-01-31T10:30:00Z"
  }
}
```

### Admin Metrics

View inbound metrics:

```bash
curl -u admin:password http://localhost:4000/api/admin/metrics
```

```json
{
  "timestamp": "2026-01-31T10:30:00Z",
  "inbound": {
    "files_discovered": 1234,
    "files_processed": 1200,
    "files_failed": 12,
    "files_pending": 22
  },
  "poller": {
    "enabled": true,
    "last_poll": "2026-01-31T10:29:00Z",
    "poll_count": 500
  }
}
```

### Telemetry Integration

The application emits telemetry events for observability:

```elixir
defmodule StaConnector.Telemetry do
  def handle_event([:sta_connector, :file, :processed], measurements, metadata, _config) do
    :telemetry.execute(
      [:sta_connector, :metrics],
      %{files_processed: 1, duration_ms: measurements.duration},
      %{system_id: metadata.system_id}
    )
  end
end
```

### Key Metrics

| Metric | Description |
|--------|-------------|
| `files_discovered` | Total files found in STA |
| `files_processed` | Successfully processed |
| `files_failed` | Failed to process |
| `files_pending` | Awaiting processing |

## Troubleshooting

### Files Not Being Discovered

1. **Check poller is enabled:**
   ```elixir
   config :sta_connector, :poller,
     enabled: true
   ```

2. **Verify system IDs are correct:**
   ```elixir
   config :sta_connector, :poller,
     systems: ["CCS", "SPI", "DICT"]
   ```

3. **Check STA connectivity:**
   ```bash
   curl http://localhost:4000/api/health
   ```

4. **Review logs:**
   ```bash
   # Check application logs
   tail -f log/prod.log | grep -i poller
   ```

5. **Check GenServer state:**
   ```elixir
   :sys.get_state(StaConnector.Files.Poller)
   ```

### Download Failures

1. **Check network connectivity to BCB**
2. **Verify Sisbacen credentials:**
   ```elixir
   config :sta_connector, :sta,
     username: "correct_user",
     password: "correct_pass"
   ```
3. **Check rate limit status** - BCB allows max 10 concurrent connections
4. **Review STA error responses in logs**

### Delivery Failures

1. **Verify destination URL is reachable:**
   ```bash
   curl -I https://staapi-planner.monetarie.com.br/api/v1/spi
   ```

2. **Check authentication credentials:**
   ```elixir
   config :sta_connector, :routes,
     SPI: %{
       headers: [{"authorization", "Bearer #{valid_token}"}]
     }
   ```

3. **Review destination server logs**

4. **Check timeout settings:**
   ```elixir
   config :sta_connector, :routes,
     SPI: %{
       timeout_ms: 60_000  # Increase if needed
     }
   ```

### High Memory Usage

1. **Reduce batch_size:**
   ```elixir
   config :sta_connector, :poller,
     batch_size: 5  # Reduced from 10
   ```

2. **Increase polling interval:**
   ```elixir
   config :sta_connector, :poller,
     interval_seconds: 60  # Increased from 30
   ```

3. **Check for stuck retries** via Admin API

4. **Monitor process memory:**
   ```elixir
   :erlang.process_info(pid, :memory)
   ```

## BCB Rate Limits

BCB enforces strict rate limits:

| Limit | Value |
|-------|-------|
| Max Concurrent Downloads | 10 |
| Max Queries per Minute | 120 |

The connector automatically enforces these limits using a semaphore pattern:

```elixir
defmodule StaConnector.RateLimiter do
  use GenServer

  @max_concurrent 10

  def acquire do
    GenServer.call(__MODULE__, :acquire, :infinity)
  end

  def release do
    GenServer.cast(__MODULE__, :release)
  end

  @impl true
  def handle_call(:acquire, from, %{current: current, queue: queue} = state) do
    if current < @max_concurrent do
      {:reply, :ok, %{state | current: current + 1}}
    else
      {:noreply, %{state | queue: :queue.in(from, queue)}}
    end
  end
end
```

If you hit rate limits:

1. Reduce `batch_size`
2. Increase `interval_seconds`
3. Monitor rate limit warnings in logs

## Next Steps

- [Outbound Pipeline](/guide/outbound) - Sending files to BCB
- [Configuration](/guide/configuration) - Full configuration reference
- [Admin API](/api/admin) - Management endpoints