# Broker ingestion

Ingestion runs in `services/broker_ingest`, one asyncio task per active broker
whose provider implements `run_ingestion`. It is off unless the config says
otherwise:

```
brokers:
  ingest_enabled: True
  ingest_processes: 1
```

`ingest_processes` sets how many OS processes supervisor starts. Process 0 runs
every broker. Higher-index processes run only providers that are safe to
replicate, which means Kafka consumers, since Kafka rebalances a consumer group
across them. REST pollers are not duplicated.

## What a survey needs

An alert becomes a candidate only if all of these are true:

1. **The broker subscribes to its topic.** Topics come from the broker's
   `altdata['kafka']['topics']`.
2. **The survey has a zeropoint.** `ZP_PER_SURVEY` in
   `skyportal/broker_apis/_save.py` maps the survey name to an AB zeropoint. A
   survey missing from it raises on every alert, which shows up in the ingest
   log. The value follows the units the points arrive in: flux in nJy is scaled
   to Jy and takes 8.9, magnitudes are converted against 23.9.
3. **Its bands map to filters.** Filters are named `<survey><band>` by default.
   A survey that names its filters otherwise needs an entry in
   `BAND_TO_FILTER_PER_SURVEY`, alongside the zeropoints.
4. **A stream carries it.** The stream's `altdata` names the collection and the
   program ids it admits.
5. **Filters are paired with the broker's own.** A `Filter` row carries the
   broker's filter id in `altdata['boom']['filter_id']`, and ingestion maps that
   back to the SkyPortal filter.

## Names do not have to match

Three names are involved and they are set independently.

* The **survey** name is what the alert carries, and it is the key into
  `ZP_PER_SURVEY`.
* The **stream** name is a SkyPortal label for the same data.
* The **topic** is the broker's, usually `<survey>_alerts_results`.

## Where the configuration lives

Broker data, including the topic list and the credentials, lives in
`Broker.altdata`, set through the API or the `/brokers/{id}` page.

## Adding a survey to a broker

1. Add `<SURVEY>` to `ZP_PER_SURVEY` if it is not already there. Use the
   zeropoint the survey's photometry is already on, so light curves stay
   continuous across a change of broker.
2. Check how its filters are named. If they are not `<survey><band>`, add the
   band mapping.
3. Add the topic to the broker's `altdata['kafka']['topics']`.
4. Check that the stream exists, and that each filter carries its partner's id
   in `altdata`.
5. Restart `broker_ingest`. Topics are read once, when the consumer starts.

## Replaying a backlog

A consumer group that has already read a topic will not read it again, and a
broker configured with `auto_offset_reset: latest` starts at the end. To take in
messages a broker has missed, add a second broker row that reads only that
topic, with its own `group_id` and `auto_offset_reset: earliest`, then
deactivate it once it has caught up.

Re-consuming an alert is safe. The save path looks for an existing candidate
with the same `passing_alert_id` before adding one, so a replay that overlaps
normal ingestion does not double up.

Retention is the limit. A backlog older than the topic's retention is gone,
whatever the offsets say.