# `DdDataStreams.Propagator`
[🔗](https://github.com/rockneurotiko/dd_data_streams/blob/main/lib/dd_data_streams/propagator.ex#L1)

The pathway header codec.

The convenience arities, `put/2` and `get/1`, use the configured key
`DdDataStreams.Config.header/0`, whose default value is
`dd-pathway-ctx-base64`, the name the official Datadog tracers use; the
explicit-key arities, `put/3` and `get/2`, use the key they are given. For
backwards compatibility with older producers,
`get/2` also falls back to the legacy binary `dd-pathway-ctx` key, decoding
its raw (non-base64) bytes, when the configured key is absent.

The payload is the 64-bit unsigned hash in
little-endian byte order followed by
the pathway start and the edge start as zigzag-encoded varints of Unix
milliseconds, with no separator or length prefix, and the whole thing is
standard base64. The timestamps are length-parsed rather than assumed to be a
fixed width, so a header written by an official tracer decodes here and a
header written here decodes there for every timestamp a real clock produces;
the widest varint the format allows diverges, and that divergence is recorded
in the interoperability fixtures. Verified against dd-trace-go v2.11.2,
dd-trace-java 1.67.1 and dd-trace-py 3.17.0.

A malformed or missing header is not an error condition for a pipeline:
`get/1` returns `nil` and the consumer starts a new pathway. `fetch/1` tells
the two apart, so a consumer that wants to report a header it could not read
can.

Next to the pathway header, a producer writes the outgoing topic under
`dd-topic` (`topic_header/0`) whenever the extractor named one. It is not part
of the pathway codec: it is the tag a consumer whose metadata names only a
subscription reads to report the queue the pathway crossed.

# `decode`

```elixir
@spec decode(binary()) :: {:ok, DdDataStreams.Pathway.t()} | :error
```

Decodes a raw header.

Bytes after the two timestamps are ignored, and a varint need not be encoded
in its shortest form. The ninth byte of a varint ends it even when its
continuation bit is set.

A start outside the plausible window makes the whole header unusable rather
than trustworthy: `get/1` then yields `nil` and the consumer starts a new
pathway, as it does for every other malformed header.

# `decode_str`

```elixir
@spec decode_str(binary()) :: {:ok, DdDataStreams.Pathway.t()} | :error
```

Decodes a base64 header string.

# `encode`

```elixir
@spec encode(DdDataStreams.Pathway.t()) :: binary()
```

Encodes a pathway as the raw header bytes.

# `encode_str`

```elixir
@spec encode_str(DdDataStreams.Pathway.t()) :: String.t()
```

Encodes a pathway as a base64 header string.

# `fetch`

```elixir
@spec fetch(map() | [tuple()] | nil) ::
  {:ok, DdDataStreams.Pathway.t()} | :absent | {:error, :malformed}
```

Reads the pathway out of a message's headers under `key`, like `get/2`, but
tells a header that is absent from one that is present and cannot be read.

Answers `{:ok, pathway}`, `:absent` when neither the configured key nor the
legacy `dd-pathway-ctx` one is present, or `{:error, :malformed}` when a
header is present but is not a usable base64 pathway: not decodable,
missing the hash and two timestamps, or carrying a start outside the window
`decode/1` believes. A consumer starts a new pathway either way, but a
producer writing a header this consumer cannot read would otherwise be
indistinguishable from one that sent no header at all. A header under any
other key is indistinguishable from an absent one.

# `fetch`

```elixir
@spec fetch(map() | [tuple()] | nil, String.t()) ::
  {:ok, DdDataStreams.Pathway.t()} | :absent | {:error, :malformed}
```

# `get`

```elixir
@spec get(map() | [tuple()] | nil) :: DdDataStreams.Pathway.t() | nil
```

Reads the pathway out of a message's headers under `key`. `get/1` defaults
to the configured key `DdDataStreams.Config.header/0`, whose default value is
`dd-pathway-ctx-base64`.

The headers are an attribute map or a list of tuples whose first element is
the header key and whose value is its last element. Returns `nil` when the
header is absent, is not a valid base64 header, does not contain a hash and
two timestamps, carries a start outside the window `decode/1` believes, or is
carried in a shape no entry can be read from.

When `key` is absent from the carrier, a legacy binary header stored under
`dd-pathway-ctx` is tried instead and decoded as raw bytes via `decode/1`.
The configured key always wins when it is present.

# `get`

```elixir
@spec get(map() | [tuple()] | nil, String.t()) :: DdDataStreams.Pathway.t() | nil
```

# `legacy_header`

```elixir
@spec legacy_header() :: String.t()
```

The legacy binary header key, `"dd-pathway-ctx"`.

# `put`

```elixir
@spec put(map() | [tuple()], DdDataStreams.Pathway.t()) :: map() | [tuple()]
```

Puts the pathway into a message's headers under `key`. `put/2` defaults to
the configured key `DdDataStreams.Config.header/0`, whose default value is
`dd-pathway-ctx-base64`.

The headers are either an attribute map or a list of tuples whose first
element is the header key, as `BroadwayKafka` exposes them; an existing entry
under the same key is replaced in both, keeping whatever other elements that
entry carries. A pathway that has not been checkpointed (hash zero) carries
no information and is not written. The headers are returned unchanged so a
produce path can call this unconditionally.

# `put`

```elixir
@spec put(map() | [tuple()], DdDataStreams.Pathway.t(), String.t()) ::
  map() | [tuple()]
```

# `topic_header`

```elixir
@spec topic_header() :: String.t()
```

The attribute or header key the producer writes the outgoing topic under,
`"dd-topic"`.

It travels next to the pathway header, so a consumer whose broker metadata
names only the subscription can still report the topic the message crossed.

---

*Consult [api-reference.md](api-reference.md) for complete listing*