# Feeds

Stream each feed's real-time events and download its hourly and daily Parquet exports.

Each feed is a `Synthient::Feed` returned by a reader on `Client`. Every feed has a real-time [stream](#streams) and hourly Parquet [exports](#exports) that roll up into daily exports. Streaming and exports need separate scopes on the API key. In the RBS signatures, `Feed` is generic over its event type, so `client.proxies` is a `Feed[ProxyEvent]`. The event types are listed under [Event types](https://docs.synthient.com/sdks/ruby/types#event-types).

| Reader         | Event type         | Stream scope            | Export scope          |
| -------------- | ------------------ | ----------------------- | --------------------- |
| `proxies`      | `ProxyEvent`       | `PROXY_FIREHOSE`        | `PROXY_FEEDS`         |
| `anonymizers`  | `AnonymizerEvent`  | `ANONYMIZERS_STREAM`    | `ANONYMIZERS_FEED`    |
| `torrents`     | `TorrentEvent`     | `TORRENTS_STREAM`       | `TORRENTS_FEED`       |
| `ja4t`         | `JA4TEvent`        | `JA4T_STREAM`           | `JA4T_FEED`           |
| `helios_http`  | `HeliosHTTPEvent`  | `HONEYPOT_HTTP_STREAM`  | `HONEYPOT_HTTP_FEED`  |
| `helios_https` | `HeliosHTTPSEvent` | `HONEYPOT_HTTPS_STREAM` | `HONEYPOT_HTTPS_FEED` |
| `helios_dns`   | `HeliosDNSEvent`   | `HONEYPOT_DNS_STREAM`   | `HONEYPOT_DNS_FEED`   |
| `helios_adb`   | `HeliosADBEvent`   | `HONEYPOT_ADB_STREAM`   | `HONEYPOT_ADB_FEED`   |

What each feed carries:

- `proxies` carries proxy sightings.
- `anonymizers` carries VPN and anonymizer address ranges.
- `torrents` carries torrents seen on the BitTorrent DHT and trackers.
- `ja4t` carries JA4T TCP fingerprint sightings.
- The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, the destinations tunnels were asked to reach, and Android Debug Bridge shell commands. Despite its name, `helios_dns` does not carry DNS queries.

See [Firehose](https://docs.synthient.com/enterprise/firehose), [Feeds](https://docs.synthient.com/enterprise/feeds), and [Helios](https://docs.synthient.com/enterprise/helios) for the data behind each feed.

## Streams

### stream

Yields the feed's real-time events as they happen, each decoded into the feed's event type `E`.

```ruby
def stream: () { (E event) -> void } -> nil
          | () -> Enumerator[E, nil]
```

- **Connection:** Each call opens a new connection. The server keeps it open for up to 30 minutes and then closes it, and `stream` returns `nil`. Breaking out of the block closes the connection.
- **Errors:** Exceptions raised in the block pass through unchanged. A response other than 200 raises an `APIError`, a dropped connection raises a `ConnectionError`, and an event that does not match the schema raises a `DecodeError`. A 429 `APIError` without `retry_after` means the key's monthly request budget is spent or it already has too many open streams.
- **Reconnecting:** The API has no resume cursor, so events sent while disconnected are missed.

Call `stream` again in a loop to reconnect:

```ruby
loop do
  client.proxies.stream do |event|
    puts "#{event.ip} #{event.provider} #{event.type}"
  end
end
```

The loop reconnects immediately after a clean close. Opening a connection already uses the SDK's bounded retries, including `Retry-After` handling. Any error that remains after those retries, or occurs while reading the stream, stops this consumer. A `429` without `Retry-After` can indicate an exhausted monthly request budget or too many open streams; resolve that condition before reconnecting.

Without a block, `stream` returns an `Enumerator` that opens the connection when iterated, so `Enumerable` methods work on it:

```ruby
sample = client.proxies.stream.first(100)
puts sample.map(&:provider).tally
```

Every feed streams the same way. Only the event type changes:

**Streams**

```ruby title="Proxies"
client.proxies.stream do |event|
  puts "#{event.ip} #{event.provider} #{event.type} #{event.country_code} #{event.asn}"
end
```

```ruby title="Anonymizers"
client.anonymizers.stream do |event|
  puts "#{event.range_start} #{event.range_end} #{event.provider} #{event.type}"
end
```

```ruby title="Torrents"
client.torrents.stream do |event|
  puts "#{event.info_hash} #{event.name} #{event.peers&.size}"
end
```

```ruby title="JA4T"
client.ja4t.stream do |event|
  puts "#{event.ip} #{event.ja4t} #{event.provider}"
end
```

```ruby title="Helios HTTP"
client.helios_http.stream do |event|
  details = event.details
  puts "#{event.domain} #{details&.http_method} #{details&.uri} #{event.meta.provider}"
end
```

```ruby title="Helios HTTPS"
client.helios_https.stream do |event|
  details = event.details
  next unless details

  puts "#{event.domain} #{details.sni} #{details.handshake_version}"
end
```

```ruby title="Helios DNS"
client.helios_dns.stream do |event|
  puts "#{event.domain} #{event.port} #{event.meta.provider}"
end
```

```ruby title="Helios ADB"
client.helios_adb.stream do |event|
  puts "#{event.session} #{event.sequential_id} #{event.command}"
end
```

`HeliosHTTPSEvent#details` is `nil` when the sensor could not parse the handshake, and `HeliosHTTPEvent#details` is optional too, so check them before use.

## Exports

Each feed has hourly Parquet exports that roll up into daily exports.

- **Hourly exports:** An hourly export is available until its day's daily export is rolled up, shortly after 00:30 UTC the next day. After that, requests for the hour raise a 400 `APIError`.
- **Daily exports:** While a daily export is being rolled up, requests for it raise a 404 `APIError` with `retry_after` set.

### Export IDs

An export ID is one of:

- `"latest"`
- a UTC day as a `Date` or `"YYYY-MM-DD"`
- a UTC hour as a `Time` or `"YYYY-MM-DD/H"`, where a zero-padded hour is also accepted

```ruby
type export_id = String | Date | Time
```

- **Latest:** `"latest"` is the newest hourly export during the day, or the daily export once that day's hours are rolled up.
- **Input:** A `Time` is converted to UTC and identifies the hour that contains it. `Export#id` values can be passed as they are. Any other string raises `ArgumentError`, and a `DateTime` or any other type raises `TypeError`, before a request is sent.
- **Helpers:** The SDK has no ID helper methods or constant for `"latest"`: pass a `Date`, a `Time`, or a string.

```ruby
yesterday = Time.now.utc.to_date - 1
last_hour = Time.now - 3600
puts client.proxies.export_meta(yesterday).rows
puts client.proxies.export_meta(last_hour).rows
```

### list\_exports

Returns one page of the feed's exports, newest first.

```ruby
def list_exports: (?limit: Integer?, ?cursor: String?) -> ExportPage
```

- `limit` `Integer?` — The page size, from 1 to 500. `nil` uses the API default of 100. A value that is not a positive `Integer` raises `ArgumentError`.
- `cursor` `String?` — Continues a listing from a previous page's `next_cursor`. `nil` or an empty string starts from the first page.

```ruby
page = client.proxies.list_exports(limit: 24)
page.exports.each do |export|
  puts "#{export.id} #{export.kind} #{export.row_count}"
end
puts "next cursor: #{page.next_cursor}"
```

- **Paging:** Use `each_export` to iterate over every page.

### each\_export

Yields every export, newest first, fetching pages as they are needed.

```ruby
def each_export: (?limit: Integer?, ?cursor: String?) { (Export export) -> void } -> nil
               | (?limit: Integer?, ?cursor: String?) -> Enumerator[Export, nil]
```

```ruby
client.proxies.each_export do |export|
  puts "#{export.id} #{export.kind} #{export.size_bytes}"
end

daily = client.proxies.each_export.find { |export| export.kind == Synthient::ExportKind::DAILY }
puts daily&.id
```

- **Paging:** Breaking out of the block stops fetching.
- **Enumerator:** Without a block, it returns an `Enumerator`.
- **Input:** `limit` and `cursor` work as they do for `list_exports`.
- **Errors:** If the API returns a cursor it already returned, `each_export` raises `Synthient::Error`.

### export\_meta

Returns the checksum, size, row count, and Parquet schema of an export.

```ruby
def export_meta: (export_id export) -> ExportMeta
```

```ruby
meta = client.proxies.export_meta("latest")
puts "#{meta.id} #{meta.rows} #{meta.size} #{meta.checksum}"
meta.schema.fields.each do |column|
  puts "#{column.name} #{column.type}"
end
```

### download\_url

Returns a presigned URL for an export's Parquet file.

```ruby
def download_url: (export_id export) -> URI::HTTP
```

```ruby
url = client.proxies.download_url("latest")
puts url.host
```

- **Lifetime:** The URL is valid for 24 hours.
- **Security:** The URL grants access without an API key, so treat it as a secret.
- **Use:** It is useful for tools that read Parquet over HTTPS directly.
- **Errors:** An error status raises an `APIError`. A success that is not a redirect, a redirect without a valid URL, or a redirect from an `https` base URL to plain `http` raises `DecodeError`.

### download

Downloads an export, yielding the Parquet file's bytes in binary chunks as they arrive.

```ruby
def download: (export_id export) { (String chunk) -> void } -> nil
            | (export_id export) -> Enumerator[String, nil]
```

- **Connection:** Breaking out of the block closes the connection.
- **Enumerator:** Without a block, it returns an `Enumerator`.
- **Security:** The API key is sent only to the Synthient API, never to the storage host that serves the file.
- **Redirects:** Up to 10 storage redirects are followed, never from `https` to `http`.
- **Errors:** A redirect the SDK cannot follow raises `Synthient::Error`. A storage response other than 200 or a redirect raises an `APIError`. Downloads from the storage host are not retried.

The SDK does not verify the file. Compare its SHA-256 with `Export#checksum` or `ExportMeta#checksum`:

```ruby
require "digest"

yesterday = Time.now.utc.to_date - 1
meta = client.proxies.export_meta(yesterday)
digest = Digest::SHA256.new
File.open("proxies.parquet", "wb") do |file|
  client.proxies.download(yesterday) do |chunk|
    file.write(chunk)
    digest << chunk
  end
end
raise "checksum mismatch" unless digest.hexdigest == meta.checksum
```