# 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.
```elixir
def deps do
[
{:teac, "~> 1.0.0"}
]
end
```
### .env
```bash
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
```elixir
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:
```elixir
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:
```elixir
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:
```elixir
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:
```elixir
{: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:
```elixir
{: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:
```elixir
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:
```elixir
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:**
```elixir
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:**
```elixir
# 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:**
```elixir
{: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:
```elixir
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:
```elixir
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:**
```elixir
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:**
```elixir
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).
```elixir
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:
```elixir
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.
```elixir
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:
```elixir
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:
```elixir
# 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:
```elixir
# 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:
```elixir
# 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:
```elixir
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:
```elixir
# 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`
```elixir
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:
```elixir
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_id`
`export TWITCH_CLIENT_SECRET=your_client_secret`
`export 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:
```elixir
{: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.
