Teac

Twitch Elixir API Client For Twitch's REST and Websocket api.

Hex docs

https://hexdocs.pm/teac/readme.html

Installation

This is currently a work in progress. While in development its highly recomended to use the github mix package as there are rapid changes coming in.

def deps do
  [
    {:teac, "~> 1.0.0"}
  ]
end

.env

export TWITCH_CLIENT_ID=""
export TWITCH_CLIENT_SECRET=""
export TWITCH_API_URI="https://api.twitch.tv/helix/"
export TWITCH_AUTH_URI="https://id.twitch.tv/oauth2/"
export TWITCH_OAUTH_CALLBACK_URI="http://example.com:4000/oauth/callbacks/twitch/"

config.ex

config :teac,
  client_id: System.get_env("TWITCH_CLIENT_ID"),
  client_secret: System.get_env("TWITCH_CLIENT_SECRET"),
  api_uri: System.get_env("TWITCH_API_URI"),
  auth_uri: System.get_env("TWITCH_AUTH_URI"),
  oauth_callback_uri: System.get_env("TWITCH_OAUTH_CALLBACK_URI")

Basic Usage

Create a client with your credentials, then pass it to any endpoint:

client = Teac.new(token: "some_auth_token", client_id: "your_client_id")

Teac.Api.Bits.Leaderboard.get(client)
Teac.Api.Chat.Color.get(client, user_id: "some_users_id")
Teac.Api.Polls.post(client, broadcaster_id: "123", title: "Best game?", choices: [...], duration: 300)

The client_id defaults to the value in your config if omitted:

client = Teac.new(token: "some_auth_token")

Check the endpoint documentation to determine whether it requires a user or app access token — some accept either, some are exclusive. https://dev.twitch.tv/docs/api/reference/

Handling Responses

All API functions return {:ok, data, rate_limit} on success or {:error, reason} on failure:

case Teac.Api.Channels.get(client, broadcaster_ids: ["123"]) do
  {:ok, [channel | _], _rl} ->
    IO.puts("Channel title: #{channel["title"]}")

  {:error, %Teac.Error{status: 401}} ->
    IO.puts("Token expired — refresh and retry")

  {:error, reason} ->
    IO.inspect(reason, label: "error")
end

Endpoints that return no body on success (e.g. PATCH, DELETE) return {:ok, nil, rate_limit}.

Rate Limits

Every response includes a %Teac.RateLimit{} as the last tuple element with three fields: limit, remaining, and reset (Unix timestamp). Use it to throttle requests or back off on 429s:

{:ok, data, rl} = Teac.Api.Users.get(client, id: "123")

if rl.remaining < 10 do
  # slow down — reset is a Unix timestamp in seconds
  Process.sleep(max(0, rl.reset * 1000 - System.os_time(:millisecond)))
end

On a 429 error the rate limit info is on the error struct:

{:error, %Teac.Error{status: 429, ratelimit: rl}} ->
  Process.sleep(max(0, rl.reset * 1000 - System.os_time(:millisecond)))

Fields are nil in test stubs where Twitch headers are absent.

Pagination

Paginated endpoints return a four-element tuple: {:ok, data, cursor, rate_limit}. Pass cursor back as :after to fetch the next page; nil cursor means you've reached the end:

def fetch_all_clips(client, broadcaster_id, cursor \\ nil, acc \\ []) do
  opts = [broadcaster_id: broadcaster_id, first: 100] ++ if cursor, do: [after: cursor], else: []

  case Teac.Api.Clips.get(client, opts) do
    {:ok, clips, nil, _rl}    -> acc ++ clips
    {:ok, clips, cursor, _rl} -> fetch_all_clips(client, broadcaster_id, cursor, acc ++ clips)
    {:error, reason}          -> {:error, reason}
  end
end

For a simpler approach, endpoints that support pagination expose a stream/2 that handles cursors automatically:

client
|> Teac.Api.Clips.stream(broadcaster_id: "123")
|> Stream.take(50)
|> Enum.to_list()

User OAuth (Authorization Code Flow)

For endpoints requiring a user access token, direct the user through Twitch's OAuth consent screen:

1. Build the authorization URL and redirect the user:

url = Teac.OAuth.AuthorizationCodeFlow.authorize_url(
  scope: [Teac.Scopes.Channel.read_subscriptions(), Teac.Scopes.Clips.edit()]
)
# redirect the user to `url`

2. Handle the callback and exchange the code for a token:

# Twitch redirects to your TWITCH_OAUTH_CALLBACK_URI with ?code=...&state=...
{:ok, %{"access_token" => token, "refresh_token" => refresh}} =
  Teac.OAuth.AuthorizationCodeFlow.exchange_code_for_token(code: params["code"])

# Store both tokens — you'll need refresh_token later
client = Teac.new(token: token)

3. Refresh when the access token expires:

{:ok, %{"access_token" => new_token, "refresh_token" => new_refresh}} =
  Teac.OAuth.AuthorizationCodeFlow.refresh_token(refresh_token: stored_refresh_token)

Twitch rotates refresh tokens on each use — always store the new refresh_token from the response.

Checking what scopes a token has:

Before redirecting to re-authorize, check whether the stored token already covers what you need:

required = [Teac.Scopes.Channel.moderate(), Teac.Scopes.User.read_chat()]

case Teac.OAuth.missing_scopes(stored_token, required) do
  {:ok, []}      -> :ok  # token already has everything
  {:ok, missing} ->
    url = Teac.OAuth.AuthorizationCodeFlow.authorize_url(scope: required)
    redirect(conn, external: url)
  {:error, :invalid_token} ->
    # token is expired — send the user through the full flow
end

missing_scopes/2 also accepts a %Teac.OAuth.TokenInfo{} from Teac.OAuth.validate_token/1 if you've already validated the token. Pass the full required scope list (not just the missing ones) to the authorize URL — Twitch won't re-prompt for already-granted scopes, but the resulting token only contains what you explicitly request.

Scopes

Teac.Scopes.* modules provide constants for all Twitch OAuth scopes, so you don't have to hardcode strings:

Teac.Scopes.Channel.read_subscriptions()  # => "channel:read:subscriptions"
Teac.Scopes.Clips.edit()                  # => "clips:edit"
Teac.Scopes.User.read_email()             # => "user:read:email"

EventSub WebSocket

Teac.WssClient connects to Twitch's EventSub WebSocket endpoint and dispatches incoming events to a handler module you provide.

1. Implement the handler behaviour:

defmodule MyApp.TwitchHandler do
  @behaviour Teac.WssClient.Handler

  @impl true
  def handle_event(%{"type" => "channel.follow"}, event, state) do
    IO.puts("New follower: #{event["user_name"]}")
    {:ok, state}
  end

  def handle_event(_subscription, _event, state), do: {:ok, state}

  @impl true
  def handle_revocation(%{"type" => type, "status" => reason}, state) do
    IO.puts("Subscription #{type} revoked: #{reason}")
    state
  end
end

2. Start the client under your supervisor:

children = [
  {Teac.WssClient, {client, handler: MyApp.TwitchHandler, name: MyApp.TwitchSocket}}
]
Supervisor.start_link(children, strategy: :one_for_one)

3. Subscribe to events after the session is established:

You must subscribe within 10 seconds of the connection or Twitch will close it (code 4003).

alias Teac.EventSub.SubscriptionTypes, as: ST

Teac.WssClient.subscribe(MyApp.TwitchSocket,
  ST.Channel.follow() ++ [condition: %{"broadcaster_user_id" => "123", "moderator_user_id" => "123"}]
)

Teac.EventSub.SubscriptionTypes.* modules provide constants that return the correct type and version for every Twitch EventSub event — merge them with your condition rather than hardcoding strings. Sub-modules cover Channel, Stream, User, Automod, Drop, Extension, and Conduit.

To remove a subscription later:

Teac.WssClient.unsubscribe(MyApp.TwitchSocket, subscription_id)

Reconnects and keepalive timeouts are handled automatically. handle_event/3 returns {:ok, new_state} — pass whatever state your handler needs through the third argument.

Conduits

Conduits are Twitch's mechanism for scaling EventSub across multiple WebSocket connections. A conduit has N shards; each shard is backed by one WebSocket session. Subscriptions are attached to the conduit rather than individual sessions, so they survive shard reconnects automatically.

Use conduits when you need more than one WebSocket connection (e.g. multiple nodes, high subscription volume). They require an app access token.

ConduitManager (recommended)

Teac.ConduitManager is a GenServer that manages the full lifecycle for you: creates the conduit, starts WebSocket sessions for each shard, assigns them, and restarts any crashed shards automatically.

children = [
  {Teac.ConduitManager, {app_client,
    shard_count: 3,
    handler: MyApp.TwitchHandler,
    name: MyApp.ConduitManager
  }}
]
Supervisor.start_link(children, strategy: :one_for_one)

Subscribe through the manager — subscriptions are durable and persist across shard reconnects:

alias Teac.EventSub.SubscriptionTypes, as: ST

Teac.ConduitManager.subscribe(MyApp.ConduitManager,
  ST.Channel.follow() ++ [condition: %{"broadcaster_user_id" => "123", "moderator_user_id" => "123"}]
)

# Remove a subscription by its Twitch subscription ID:
Teac.ConduitManager.unsubscribe(MyApp.ConduitManager, subscription_id)

ConduitManager reuses an existing conduit on startup if one is already registered for your app token, so restarts don't create duplicate conduits.

Low-level conduit API

If you need direct control, the REST and WebSocket primitives are available individually:

# Create a conduit
{:ok, [%{"id" => conduit_id}], _rl} =
  Teac.Api.EventSub.Conduits.post(app_client, shard_count: 1)

# Connect a WebSocket session and assign it to a shard (within 10 seconds of welcome)
{:ok, pid} = Teac.WssClient.start_link(app_client, handler: MyApp.TwitchHandler)
Teac.WssClient.assign_shard(pid, conduit_id: conduit_id, shard_id: "0")

# Create subscriptions via REST, pointing at the conduit
Teac.Api.EventSub.Subscriptions.post(app_client,
  payload: %{
    "type" => "channel.follow",
    "version" => "2",
    "condition" => %{"broadcaster_user_id" => "123", "moderator_user_id" => "123"},
    "transport" => %{"method" => "conduit", "conduit_id" => conduit_id}
  }
)

# Check shard status
{:ok, shards, _cursor, _rl} =
  Teac.Api.EventSub.Conduits.Shards.get(app_client, conduit_id: conduit_id)

EventSub Webhooks

Teac.Plug.EventSub receives EventSub events via HTTP POST instead of WebSocket. It verifies Twitch's HMAC-SHA256 signature on every request, handles the one-time challenge handshake, and dispatches to the same Teac.WssClient.Handler behaviour — so you can reuse your handler across both transports.

Phoenix setup:

Add the route before your main pipeline so the body is not consumed before signature verification:

# router.ex
scope "/webhooks" do
  forward "/twitch", Teac.Plug.EventSub,
    secret: "your-webhook-secret",
    handler: MyApp.TwitchHandler,
    handler_state: %{}
end

Because Phoenix's Plug.Parsers reads the body, configure Teac.Plug.CacheBodyReader so the raw bytes remain available for HMAC verification:

# endpoint.ex
plug Plug.Parsers,
  parsers: [:urlencoded, :multipart, :json],
  pass: ["*/*"],
  body_reader: {Teac.Plug.CacheBodyReader, :read_body, []},
  json_decoder: JSON

Creating webhook subscriptions:

Webhook subscriptions are created via the REST API with "webhook" as the transport method. Twitch will POST to your endpoint URL immediately to verify the challenge, then deliver events:

Teac.Api.EventSub.Subscriptions.post(app_client,
  payload: %{
    "type" => "channel.follow",
    "version" => "2",
    "condition" => %{"broadcaster_user_id" => "123", "moderator_user_id" => "123"},
    "transport" => %{
      "method" => "webhook",
      "callback" => "https://yourapp.com/webhooks/twitch",
      "secret" => "your-webhook-secret"
    }
  }
)

Deduplication:

Twitch retries failed deliveries up to three times with the same Twitch-Eventsub-Message-Id. The plug does not deduplicate — if your handler performs non-idempotent operations, check the message ID yourself before processing:

# In a custom plug wrapping Teac.Plug.EventSub, or inside handle_event/3:
msg_id = List.first(Plug.Conn.get_req_header(conn, "twitch-eventsub-message-id"))
# check msg_id against your store; skip if already seen

Twitch stops retrying once it receives any 2xx response.

Example Application using this lib.

https://codeberg.org/FullStacking/teac_example

Using App Flow Auth token.

A genserver that fetches and mantains a valid auth token for Client Credential is provided. IE: Server To Server or noted as on any endpoint as Requires an app access token

def start(_type, _args) do
  children = [
    ...
    Teac.Oauth.ClientCredentialManager,
    ...
  ]
  opts = [strategy: :one_for_one, name: TeacExample.Supervisor]
  Supervisor.start_link(children, opts)
end

Assuming you provided your .env vars and have a running server, retrieve a token and build a client:

token = Teac.Oauth.ClientCredentialManager.get_token()
client = Teac.new(token: token)

get_token/0 always returns a valid app token. Pass the resulting client to any endpoint that accepts an app access token.

Developing

You will need the twitch cli tool. https://dev.twitch.tv/docs/cli/

• Generate some mock data twitch mock-api generate -c 6

• Set Env Variables from mock output export TWITCH_CLIENT_ID=your_client_idexport TWITCH_CLIENT_SECRET=your_client_secretexport TWITCH_API_URI="http://localhost:8080"

• Start a mock twitch server. twitch mock-api serve

Assuming you have a Twitch mock server running to get the auth:

{:ok, [%{"ID" => client_id, "Secret" => client_secret}]} = Teac.MockApi.clients()

{:ok, %{"access_token" => access_token}} =
  Teac.MockAuth.fetch_app_access_token(client_id: client_id, client_secret: client_secret)

client = Teac.new(token: access_token, client_id: client_id)
{:ok, data, _rl} = Teac.Api.Bits.Cheermotes.get(client)

All endpoints take a %Teac.Client{} as the first argument, followed by an optional keyword list of endpoint-specific parameters.