# 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. Flux in nJy is scaled to Jy and takes 8.9, and magnitudes are converted
   against the same zeropoint.
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

Four 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 **instrument** is the one named after the survey, unless
  `INSTRUMENT_PER_SURVEY` in `skyportal/utils/survey.py` maps it to another
  (`DECAM` is stored under `DECam`).
- 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.