# Feeds

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

Each feed is a field of `Client` with type `*Feed[T]`, where `T` is the feed's event type. 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. The event types are listed under [Event types](https://docs.synthient.com/sdks/go/types#event-types).

| Field         | 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`           |
| `HeliosHTTP`  | `HeliosHTTPEvent`  | `HONEYPOT_HTTP_STREAM`  | `HONEYPOT_HTTP_FEED`  |
| `HeliosHTTPS` | `HeliosHTTPSEvent` | `HONEYPOT_HTTPS_STREAM` | `HONEYPOT_HTTPS_FEED` |
| `HeliosDNS`   | `HeliosDNSEvent`   | `HONEYPOT_DNS_STREAM`   | `HONEYPOT_DNS_FEED`   |
| `HeliosADB`   | `HeliosADBEvent`   | `HONEYPOT_ADB_STREAM`   | `HONEYPOT_ADB_FEED`   |

What each feed carries:

- `Proxies` carries residential, mobile, and datacenter proxy sightings.
- `Anonymizers` carries VPN and anonymizer IP ranges.
- `Torrents` carries torrents seen on the BitTorrent DHT and trackers.
- `JA4T` carries JA4T TCP fingerprints.
- The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, tunnel destinations, and Android Debug Bridge commands. Despite its name, `HeliosDNS` 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

Returns an iterator over the feed's real-time events.

```go
func (f *Feed[T]) Stream(ctx context.Context) iter.Seq2[T, error]
```

- **Connection:** Each `range` over the iterator opens a new connection, which the server keeps open for up to 30 minutes and then closes cleanly. Breaking out of the loop closes the connection.
- **Errors:** A clean close ends the iteration without an error. Any other failure, including a non-success response (as an `*APIError`) or a canceled `ctx`, is yielded once and ends the iteration. A 429 `*APIError` without `RetryAfter` 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 between connections are missed.

Range over `Stream` again to reconnect:

```go
for ctx.Err() == nil {
	for event, err := range client.Proxies.Stream(ctx) {
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println(event.IP, event.Provider, event.Type)
	}
}
```

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.

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

**Streams**

```go title="Proxies"
for event, err := range client.Proxies.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.IP, event.Provider, event.Type, event.CountryCode, event.ASN)
}
```

```go title="Anonymizers"
for event, err := range client.Anonymizers.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.RangeStart, event.RangeEnd, event.Provider, event.Type)
}
```

```go title="Torrents"
for event, err := range client.Torrents.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.InfoHash, event.Name, len(event.Peers))
}
```

```go title="JA4T"
for event, err := range client.JA4T.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.IP, event.JA4T, event.Provider)
}
```

```go title="Helios HTTP"
for event, err := range client.HeliosHTTP.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.Domain, event.Details.Method, event.Details.URI, event.Meta.Provider)
}
```

```go title="Helios HTTPS"
for event, err := range client.HeliosHTTPS.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	if event.Details == nil {
		continue
	}
	fmt.Println(event.Domain, event.Details.SNI, event.Details.HandshakeVersion)
}
```

```go title="Helios DNS"
for event, err := range client.HeliosDNS.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.Domain, event.Port, event.Meta.Provider)
}
```

```go title="Helios ADB"
for event, err := range client.HeliosADB.Stream(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(event.Session, event.SequentialID, event.Command)
}
```

`HeliosHTTPSEvent.Details` is `nil` when the sensor could not parse the handshake, so check it 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 return a 400 `*APIError`.
- **Daily exports:** While a daily export is being rolled up, requests for it return a 404 `*APIError` with `RetryAfter` set.

### Export IDs

An export ID is one of:

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

`Export.ID` values can be passed as they are. Any other string returns an error before a request is sent.

- `LatestExport` `const string` — `"latest"`, the feed's freshest export: the newest hourly export during the day, or the daily export once that day's hours are rolled up.
- `DailyExport` `func(t time.Time) string` — Returns the ID of the daily export for the UTC day containing `t`.
- `HourlyExport` `func(t time.Time) string` — Returns the ID of the hourly export for the UTC hour containing `t`.

```go
yesterday := synthient.DailyExport(time.Now().AddDate(0, 0, -1))
lastHour := synthient.HourlyExport(time.Now().Add(-time.Hour))
fmt.Println(yesterday, lastHour)
```

### ListExports

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

```go
func (f *Feed[T]) ListExports(ctx context.Context, params ...ListExportsParams) (*ExportPage, error)
```

- `ListExportsParams.Limit` `int` — The page size, from 1 to 500. Zero uses the API default of 100.
- `ListExportsParams.Cursor` `string` — Continues a listing from a previous page's `NextCursor`.

```go
page, err := client.Proxies.ListExports(ctx, synthient.ListExportsParams{Limit: 24})
if err != nil {
	log.Fatal(err)
}
for _, export := range page.Exports {
	fmt.Println(export.ID, export.Kind, export.RowCount)
}
fmt.Println("next cursor:", page.NextCursor)
```

- **Input:** `params` is optional, and only the first is used.
- **Paging:** Use `AllExports` to iterate over every page.

### AllExports

Returns an iterator over the feed's exports, newest first.

```go
func (f *Feed[T]) AllExports(ctx context.Context, params ...ListExportsParams) iter.Seq2[Export, error]
```

```go
for export, err := range client.Proxies.AllExports(ctx) {
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(export.ID, export.Kind, export.SizeBytes)
}
```

- **Paging:** It fetches pages lazily as iteration proceeds, and stops without fetching more when the loop breaks.
- **Errors:** An error is yielded once and ends the iteration, including when the API returns a cursor it already returned.
- **Input:** `params` is optional, and only the first is used.

### ExportMeta

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

```go
func (f *Feed[T]) ExportMeta(ctx context.Context, id string) (*ExportMeta, error)
```

```go
meta, err := client.Proxies.ExportMeta(ctx, synthient.LatestExport)
if err != nil {
	log.Fatal(err)
}
fmt.Println(meta.ID, meta.Rows, meta.Size, meta.Checksum)
for _, column := range meta.Schema.Fields {
	fmt.Println(column.Name, column.Type)
}
```

### DownloadURL

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

```go
func (f *Feed[T]) DownloadURL(ctx context.Context, id string) (*url.URL, error)
```

```go
u, err := client.Proxies.DownloadURL(ctx, synthient.LatestExport)
if err != nil {
	log.Fatal(err)
}
fmt.Println(u.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.

### Download

Returns the contents of an export's Parquet file.

```go
func (f *Feed[T]) Download(ctx context.Context, id string) (io.ReadCloser, error)
```

- **Cleanup:** The caller must close the returned reader.
- **Security:** The API key is sent only to the Synthient API, never to the storage host that serves the file, and errors never include the presigned URL.
- **Errors:** A non-success response from either returns 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`:

```go
func saveExport(ctx context.Context, feed *synthient.Feed[synthient.ProxyEvent], id, path string) error {
	meta, err := feed.ExportMeta(ctx, id)
	if err != nil {
		return err
	}
	body, err := feed.Download(ctx, id)
	if err != nil {
		return err
	}
	defer body.Close()

	file, err := os.Create(path)
	if err != nil {
		return err
	}
	hash := sha256.New()
	_, err = io.Copy(io.MultiWriter(file, hash), body)
	err = errors.Join(err, file.Close())
	if err != nil {
		return err
	}

	if hex.EncodeToString(hash.Sum(nil)) != meta.Checksum {
		return errors.New("checksum mismatch")
	}
	return nil
}
```

```go
yesterday := synthient.DailyExport(time.Now().AddDate(0, 0, -1))
err := saveExport(ctx, client.Proxies, yesterday, "proxies.parquet")
if err != nil {
	log.Fatal(err)
}
```

Call it with a fixed ID rather than `LatestExport`, which can move to a newer export between the `ExportMeta` and `Download` requests.