Rindle.Streaming.Provider.Mux (Rindle v0.1.5)

Copy Markdown View Source

Mux REST adapter implementing Rindle.Streaming.Provider.

Configuration

Credentials and tunables are resolved from the application environment on every call (D-30 — no caching at module load time; adopters using config/runtime.exs are unaffected):

config :rindle, Rindle.Streaming.Provider.Mux,
  token_id: System.get_env("RINDLE_MUX_TOKEN_ID"),
  token_secret: System.get_env("RINDLE_MUX_TOKEN_SECRET"),
  signing_key_id: System.get_env("RINDLE_MUX_SIGNING_KEY_ID"),
  signing_private_key: System.get_env("RINDLE_MUX_SIGNING_PRIVATE_KEY"),
  webhook_secrets:
    System.get_env("RINDLE_MUX_WEBHOOK_SECRETS", "")
    |> String.split(",", trim: true),
  webhook_tolerance_seconds: 300,
  provider_polling_floor_seconds: 30,
  provider_stuck_threshold_seconds: 7200

DSL ↔ Mux REST translation (D-04 memo correction)

Phase 33 ships the DSL atom :playback_policy (singular) and the schema column playback_policy (singular). The Phase 33 schema also defines field :playback_ids, {:array, :string} (PLURAL ARRAY).

At the SDK boundary this adapter translates to the current Mux REST API keys: inputs (PLURAL list of objects) and playback_policies (PLURAL string list). The Mux singular keys are deprecated as of 2026-05 — always use plural here. Param construction lives in a single private helper (build_create_params/2) so workers never duplicate this logic.

The Phase 33 callback contract returns playback_ids: [playback_id()] (a list); the row schema persists playback_ids as {:array, :string}. The adapter writes the array verbatim; reads use List.first/1 only when a single id is needed (e.g., for URL minting).

Test wiring

Tests configure the adapter to route HTTP calls through Rindle.Streaming.Provider.Mux.ClientMock by overriding the :http_client key on the same :rindle, __MODULE__ config keyspace (see test/rindle/streaming/provider/mux/mux_test.exs for the canonical setup). The default :http_client is Rindle.Streaming.Provider.Mux.HTTP.

Security invariants

  • provider_asset_id never crosses into adopter-facing URLs, log lines, or telemetry metadata. Only the public-side playback_id is embedded in URLs. Telemetry emit sites use Rindle.Domain.MediaProviderAsset.redact_id/1 (security invariant 14).
  • signed_playback_url/3 ALWAYS passes :expiration explicitly to Mux.Token.sign_playback_id/2. The SDK default is 7 days (Pitfall 1) — relying on it would silently mint over-long tokens.

Telemetry Contract

The adapter and its companion workers emit the following events. All events with metadata.asset_id redact the value to its last-4-char tag ("...abcd") via Rindle.Domain.MediaProviderAsset.redact_id/1 (security invariant 14). Adopters writing telemetry handlers MUST treat asset_id as a redacted identifier — it is not a stable correlation key.

  • [:rindle, :provider, :ingest, :start | :stop | :exception] — emitted by Rindle.Workers.MuxIngestVariant (Phase 34).

    • measurements: %{system_time, duration?} (duration only on :stop/:exception)
    • metadata: %{profile, provider, asset_id, variant_name, kind?, reason?}

    • kind: :error | :cancelled is added on :exception to distinguish genuine errors ({:error, _}) from atomic-promote cancellations ({:cancel, {:stale_source, _}}).

  • [:rindle, :provider, :sync, :resolved | :stuck] — emitted by Rindle.Workers.MuxSyncProviderAsset (Phase 34).

    • measurements: %{system_time}
    • metadata: %{profile, provider, asset_id, provider_state, age_ms}
    • :stuck fires when a row in :processing/:uploading exceeds :provider_stuck_threshold_seconds (default 7200).

  • [:rindle, :delivery, :streaming, :resolved] — already shipped by Phase 33's dispatch_streaming/4. No new event from Phase 34 on this path; the metadata extension is the documented v1.4-contract addition.

Phase 35 will add [:rindle, :provider, :webhook, _] events. Phase 34 does not emit them.

Summary

Functions

create_asset(profile, source_url, opts \\ [])

Server-push ingest entry point. Translates DSL :playback_policy (singular, via opts) to the Mux REST PLURAL playback_policies key. Returns the Phase 33 contract shape {:ok, %{provider_asset_id: _, playback_ids: [_]}} (PLURAL array, even with a single element).

create_asset_with_retry_hint(profile, source_url, opts \\ [])

Worker-facing variant of create_asset/3 that exposes the 429 Retry-After seconds value so the Plan 02 worker can snooze cleanly. Param construction (PLURAL keys) lives ONLY here in the adapter — never duplicated in workers.

Functions

create_asset(profile, source_url, opts \\ [])

Server-push ingest entry point. Translates DSL :playback_policy (singular, via opts) to the Mux REST PLURAL playback_policies key. Returns the Phase 33 contract shape {:ok, %{provider_asset_id: _, playback_ids: [_]}} (PLURAL array, even with a single element).

Errors are normalized to the Phase 33 atom set:

  • :provider_quota_exceeded (HTTP 429) — caller can extract Retry-After from %Tesla.Env{}.headers via create_asset_with_retry_hint/3 if it needs the snooze duration (Plan 02 worker uses that variant).
  • :provider_sync_failed (HTTP 4xx/5xx other than 429).

create_asset_with_retry_hint(profile, source_url, opts \\ [])

@spec create_asset_with_retry_hint(module(), String.t(), keyword()) ::
  {:ok, %{provider_asset_id: String.t(), playback_ids: [String.t()]}}
  | {:error, :provider_quota_exceeded, non_neg_integer()}
  | {:error, atom()}
  | {:error, term()}

Worker-facing variant of create_asset/3 that exposes the 429 Retry-After seconds value so the Plan 02 worker can snooze cleanly. Param construction (PLURAL keys) lives ONLY here in the adapter — never duplicated in workers.

Returns:

  • {:ok, %{provider_asset_id: _, playback_ids: [_]}} — happy path.
  • {:error, :provider_quota_exceeded, retry_after_seconds} — HTTP 429 with parsed Retry-After (60-second floor when header is missing or unparseable).
  • {:error, :provider_sync_failed} — other 4xx/5xx.
  • {:error, term()} — transport/lower-level error.