liquidsoap.liq

# WARNING! This file is automatically generated by AzuraCast.
# Do not update it directly!

# Custom Configuration (Specified in Station Profile)
### BEGIN: BFF.fm Configuration and Secrets ###
### Secrets should not be checked in to source control. Duh. ###

# General Config

## User Agent: We generate a user agent for HTTP requests using these
user_agent_name = ref("BFF.fm Broadcast System")
user_agent_url = ref("https://bff.fm")

## Fixed MP3 source content
# Our error/info default recordings are stored in the cloud, you can also reference
# local files.
mp3_technical_difficulties = ref("https://.../broadcast-server-test-feed.mp3")
mp3_test_feed = ref("https://.../broadcast-server-test-feed.mp3")

## Azuracast Public API info
azuracast_public_url = ref("https://broadcast.example.com")
azuracast_public_api_key = ref("(PASSWORD)")

## Creek CMS integration
# BFF.fm uses a modified version of Creek as its CMS, through which we authenticate
# our streamers, and also pull canonical track metadata.
creek_app_id = ref("fm.bff.example")
creek_api_base = ref("https://")
creek_public_base = ref("https://")

## Ingestion harbor configuration
# To make re-use and hiding passwords in the checked-in version simpler, set these
# Icecast output variables.
icecast_port = 8005
icecast_password = ref("(PASSWORD)")

## Fixed passwords for studio Icecast stream harbors
# Our studio connections use a single client ID and so have fixed passwords hard-coded
# into this config, rather than authenticating against Creek.
ingest_studio_a_password = ref("(PASSWORD)")
ingest_fallback_password = ref("(PASSWORD)")
ingest_emergency_password = ref("(PASSWORD)")

## Last.FM scrobbling
# We scrobble everything we play when metadata is pulled from Creek.
lastfm_app_id = ref("(PASSWORD)")
lastfm_app_secret = ref("(PASSWORD)")
# You'll need to do an OAuth dance with the app credentials above to get
# a session token for your Last.FM account to scrobble. You can find a node
# utility script to do that here: https://github.com/BFFdotFM/logstash-scrobbler
lastfm_session_token = ref("(PASSWORD)") # @bffdotfm

## Slack logging
# In some world, we'd probably log more actively to a real logging platform,
# but as a community radio station Slack is very, very useful for basic visibility
# and this config throws info like new connections/failed authentication into
# specified slack channels.

# Prefix all Slack logs with this text (useful to distinguish debugging instances from
# your production messages.)
# e.g. :safety_vest:, [DEBUG], etc.
slack_message_prefix = ref(":safety_vest: ")

# We post to Slack using webhooks, so you'll need to create a Slack app for your instance,
# and then create webhooks for the channels you want to post to. Calls to `slack.log` can
# be named, allowing different channels to receive messages (e.g. `alerts`, `stream_auth`,
# and `now_playing`.)
#
# Each assignment only needs to unique path of the webhook, so will look something like
# `ref("J12345678/G1234567890/1234567890abcdef12345678")`
slack_default_channel = ref("(PASSWORD)")
slack_named_channels = ref([
  ("alerts", slack_default_channel),
  ("stream_auth", slack_default_channel),
  ("now_playing", ref("(PASSWORD)"))
])

## TuneIn Air
# As with Last.FM, you'll need to register for the TuneIn Air service to send first-class
# tracking info to TuneIn, and set the various keys here:
tunein_partner_id = ref("(PASSWORD)")
tunein_partner_key =  ref("(PASSWORD)")
tunein_station_id = ref("(PASSWORD)")

### END BFF.fm Configuration and Secrets ###

init.daemon.set(false)
init.daemon.pidfile.path.set("/var/azuracast/stations/example/config/liquidsoap.pid")

log.stdout.set(true)
log.file.set(false)

settings.server.log.level.set(4)

settings.server.socket.set(true)
settings.server.socket.permissions.set(0o660)
settings.server.socket.path.set("/var/azuracast/stations/example/config/liquidsoap.sock")

settings.harbor.bind_addrs.set(["0.0.0.0"])

settings.tag.encodings.set(["UTF-8","ISO-8859-1"])
settings.encoder.metadata.export.set(["artist","title","album","song"])

setenv("TZ", "America/Los_Angeles")

autodj_is_loading = ref(true)
ignore(autodj_is_loading)

autodj_ping_attempts = ref(0)
ignore(autodj_ping_attempts)

# Track live-enabled status.
live_enabled = ref(false)
ignore(live_enabled)

# Track live transition for crossfades.
to_live = ref(false)
ignore(to_live)

azuracast_api_url = "http://127.0.0.1:6010/api/internal/1/liquidsoap"
azuracast_api_key = "(PASSWORD)"

def azuracast_api_call(~timeout_ms=2000, url, payload="") =
    full_url = "#{azuracast_api_url}/#{url}"

    log("API #{url} - Sending POST request to '#{full_url}' with body: #{payload}")
    try
        response = http.post(full_url,
            headers=[
                ("Content-Type", "application/json"),
                ("User-Agent", "Liquidsoap AzuraCast"),
                ("X-Liquidsoap-Api-Key", "#{azuracast_api_key}")
            ],
            timeout_ms=timeout_ms,
            data=payload
        )

        log("API #{url} - Response (#{response.status_code}): #{response}")
        "#{response}"
    catch err do
        log("API #{url} - Error: #{error.kind(err)} - #{error.message(err)}")
        "false"
    end
end

station_media_dir = "/var/azuracast/stations/example/media"
def azuracast_media_protocol(~rlog=_,~maxtime=_,arg) =
    ["#{station_media_dir}/#{arg}"]
end

add_protocol(
    "media",
    azuracast_media_protocol,
    doc="Pull files from AzuraCast media directory.",
    syntax="media:uri"
)

# Custom Configuration (Specified in Station Profile)
### BEGIN: BFF.fm Settings Overrides ###

# log level 5 for mega debugging
# settings.server.log.level.set(5)

### END: BFF.fm Settings Overrides


### BEGIN: BFF.fm Library Code ###
## After: Azuracast helpers, `add_protocol("media", ...)`

## Time Utilities
# Shorthands for various time formats

let timestamp = ()

# Default to the system timestamp and 'now'
def timestamp.custom(format="%T", datetime="now") =
  list.hd(default="", process.read.lines(env=[("TZ", ":America/Los_Angeles"), ("FORMAT", format), ("DT", datetime)], "date -d\"$DT\" +\"$FORMAT\""))
end

# e.g. 20:45:23
def timestamp.hms(datetime="now") =
  timestamp.custom("%T", datetime)
end

# e.g. 2022-09-14T03:52:38
def timestamp.iso8609(datetime="now") =
  timestamp.custom("%Y-%m-%dT%H:%M:%S", datetime)
end

# e.g. 20220914T035238
def timestamp.ymdhms(datetime="now") =
  timestamp.custom("%Y%m%dT%H%M%S", datetime)
end

# e.g. 1663131225
def timestamp.unix(datetime="now") =
  timestamp.custom("%s", datetime)
end

# e.g. 9:49pm on 13 September, 2022
def timestamp.friendly(datetime="now") =
  timestamp.custom("%-I:%M%P on %-d %B, %Y", datetime)
end

## Hash and Crypto
# Last.FM's API needed some MD5 smushing, so here it is.
let hash = ()
def hash.md5(str) =
  md5sum = file.which("md5sum")
  log.debug("Path to md5sum: #{md5sum}")
  # md5sum returns the form "<HASH> -", so split away just the hash w/ awk:
  list.hd(default="", process.read.lines(env=[("STR", str)], "printf %s \"$STR\" | #{md5sum} | awk '{ print $1 }'"))
end

## Curl HTTP functions
# While Azuracast w/ Liquidsoap 2 now has curllib HTTP functions build in (these
# function used to shell out to curl), we still use these wrappers provide some
# sugar around the build-in http.get/post methods, particularly for data wrangling
let curl = ()
let curl.version = ref(list.hd(default="curl/unknown", process.read.lines("curl --version | head -1 | awk '{ print $1 \"/\" $2 }'")))
let curl.userAgent = ref("Liquidsoap/#{liquidsoap.version} (#{!curl.version}) #{!user_agent_name} <#{!user_agent_url}>")

def curl.headers(headers=[]) =
  list.add(("User-Agent", !curl.userAgent), headers)
end

# Provide a data object ala
#
# {
#   key: "data",
#   another: "more"
# }
#
def curl.json(~data={}, ~timeout_ms=2000, ~headers=[], theUrl) =
  log("cURL JSON: #{theUrl}")
  try
    response = http.post(theUrl,
      headers=curl.headers(list.add(
        ("Content-Type", "application/json"),
        headers
      )),
      timeout_ms=timeout_ms,
      data=json.stringify(compact=true, data)
    )

    log("cURL JSON: #{theUrl} - Response (#{response.status_code})")
    "#{response}"
  catch err do
    log("cURL JSON: #{theUrl} - Error: #{error.kind(err)} - #{error.message(err)}")
    "Unknown HTTP error"
  end
end

# Provide the data object as a list of pairs
#
# [
#   ("key", "data"),
#   ("another", "more")
# ]
#
def curl.post(~data=[], ~timeout_ms=2000, ~headers=[], theUrl) =
  log("cURL POST: #{theUrl}")

  def buildDataString(data, arg) =
    data ^ fst(arg) ^ "=" ^ url.encode(snd(arg)) ^ "&"
  end
  formData = list.fold(buildDataString, "", data)

  try
    response = http.post(theUrl,
      headers=curl.headers(list.add(
        ("Content-Type", "application/x-www-form-urlencoded"),
        headers
      )),
      timeout_ms=timeout_ms,
      data=formData
    )
    log("cURL POST: #{theUrl} - Response (#{response.status_code})")
    "#{response}"
  catch err do
    log("cURL POST: #{theUrl} - Error: #{error.kind(err)} - #{error.message(err)}")
    "Unknown HTTP error"
  end
end

def curl.fetch(~timeout_ms=2000, ~headers=[], theUrl) =
  log("cURL GET: #{theUrl}")
  try
    response = http.get(theUrl,
      headers=curl.headers(list.add(
        ("Content-Type", "application/x-www-form-urlencoded"),
        headers
      )),
      timeout_ms=timeout_ms
    )
    "#{response}"
  catch err do
    log("cURL GET: #{theUrl} - Error: #{error.kind(err)} - #{error.message(err)}")
    "Unknown HTTP error"
  end
end

## Azuracast Public API Functions
# Functionality that calls the public API for AzuraCast, rather than the internal
# azuracast_api_call version.
#
# You need to provide an API key with `Administer Stations` permission.
let azuracast_public = ()
let azuracast_public.apiBase = azuracast_public_url
let azuracast_public.apiKey = azuracast_public_api_key
  
# Extract station ID from internal API url above: `/internal/{number}/`
def azuracast_public.stationId() = 
  let matcher = regexp("/internal/(\\d+)/")
  # returns [(index, value)]
  snd(list.hd(matcher.exec(azuracast_api_url), default=(0, "0")))
end
  
def azuracast_public.call(endpoint, payload={}) =
  let api_url = "#{!azuracast_public.apiBase}/api/station/#{azuracast_public.stationId()}/#{endpoint}"
  log("Azuracast Public: #{api_url}")

  ignore(curl.json(
    data=payload,
    headers=[
      ("X-Api-Key", !azuracast_public.apiKey)
    ],
    api_url))
end

## BFF.fm API functions
#
# The BFF.fm CMS is a forked version of Creek.fm, along with some of our own
# deployment idiocyncracies, so this wrapper makes calling the API easier
let bff = ()
let bff.appId = creek_app_id
let bff.base = creek_api_base
let bff.publicBase = creek_public_base

def bff.fetch(~cacheBust=false, endpoint)
  appendUrl = if (cacheBust == true) then
    "&cacheBust=#{timestamp.unix()}"
  else
    ""
  end

  fullApiCallUrl = "#{!bff.base}/api/#{endpoint}?app_id=#{url.encode(!bff.appId)}#{appendUrl}"
  curl.fetch(fullApiCallUrl)
end

def bff.post(~data=[], endpoint)
  url = "#{!bff.base}/api/#{endpoint}"
  fullData = list.append([
    ('app_id', !bff.appId)
  ], data)
  curl.post(data=fullData, url)
end

# Create a special "now" endpoint shortlink that will direct users to the most relevant show/broadcast
# destination based on the time of the URL:
#
# e.g. https://bff.fm/now/20200209T115500
def bff.nowUrl() =
  "#{creek_public_base}/now/#{timestamp.ymdhms()}"
end

## Slack Functions
# BFF.fm makes heavy use of Slack for all our community organizing and a lot of technical monitoring.
let slack = ()

# These options are documented in the configuration block above
let slack.defaultChannel = slack_default_channel
let slack.channels = slack_named_channels
let slack.messagePrefix = slack_message_prefix

def slack.send(hook, message) =
  messageObject = {
    text = "[#{timestamp.hms()}] #{!slack.messagePrefix}#{message}",
    type = "mrkdwn"
  }
  log("SLACK: #{message}")
  ignore(curl.json(data=messageObject, "https://hooks.slack.com/services/#{hook}"))
end

def slack.log(~channel="", message) =
  channelHook = list.assoc(default=slack.defaultChannel, channel, !slack.channels)
  slack.send(!channelHook, message)
end

## Last.FM Functions
#
# Last.FM integration can also be provided via an extension for Liquidsoap, but it's
# not shipping in AzuraCast at this time, so we implemented the `track.scrobble` API
# ourselves.
#
# You'll need to create an app at https://www.last.fm/api and then auth your account
# against it to get a Session token and store it all in the config above.
let lastfm = ()
let lastfm.apiBase = ref("https://ws.audioscrobbler.com/2.0/")
let lastfm.appId = lastfm_app_id
let lastfm.appSecret = lastfm_app_secret
let lastfm.sessionToken = lastfm_session_token

def lastfm.scrobble(~artist, ~track, ~album="") =

  log.info("Scrobbling: #{track} by #{artist} (Album: #{album})")

  # parameters in alphabetical order
  params = [
    ("album", album),
    ("api_key", !lastfm.appId),
    ("artist", artist),
    ("method", "track.scrobble"),
    ("sk", !lastfm.sessionToken),
    ("timestamp", timestamp.unix()),
    ("track", track),
  ]

  def flattenSignatureParam(output, p) =
    output ^ fst(p) ^ snd(p)
  end

  hashableParamString = list.fold(flattenSignatureParam, "", params) ^ !lastfm.appSecret
  apiSignature = hash.md5(hashableParamString)
  fullParams = list.append(params, [("api_sig", apiSignature)])
  ignore(curl.post(data=fullParams, !lastfm.apiBase))
end

# While we now pull our playing data from a richer JSON API, this function handles the common
# Artist - Track string format that might be passed around your automation system.
def lastfm.scrobbleNowPlayingString(nowPlaying) =
  parts = list.map(string.trim, string.split(separator="-", nowPlaying))

  if (list.length(parts) == 2) then
    artist = list.hd(default="", parts)
    track = list.nth(default="", parts, 1)
    lastfm.scrobble(artist=artist, track=track)
  else
    log.important("Could not attempt scrobbling #{nowPlaying}, did not split into 2 parts")
  end
end

## TuneIn AIR Support
# AzuraCast does have some native support for TuneIn, but since we're pulling richer track
# metadata (see below), we've implemented it directly.

let tunein = ()
let tunein.apiBase = ref("http://air.radiotime.com/Playing.ashx")
let tunein.partnerId = tunein_partner_id
let tunein.partnerKey = tunein_partner_key
let tunein.stationId = tunein_station_id

def tunein.airRequest(params) =
  fullParams = list.append([
    ("partnerId", !tunein.partnerId),
    ("partnerKey", !tunein.partnerKey),
    ("id", !tunein.stationId)
  ], params)
  ignore(curl.post(data=fullParams, !tunein.apiBase))
end

def tunein.nowPlaying(~artist, ~track, ~album="") =
  log.info("Logging to TuneIn AIR NOW: #{track} by #{artist} (Album: #{album})")
  tunein.airRequest([
    ("artist", artist),
    ("title", track),
    ("album", album)
  ])
end

def tunein.nonMusic(~attribution, ~name) =
  log.info("Logging non-music to TuneIn AIR NOW: #{name} attributed to #{attribution}")
  tunein.airRequest([
    ("artist", attribution),
    ("title", name),
    # ("commercial", "true")
  ])
end

### END: BFF.fm Library Code ###

playlist_silence = playlist(id="playlist_silence",mime_type="audio/x-mpegurl",mode="normal",reload_mode="watch","/var/azuracast/stations/ebbfs/playlists/playlist_silence.m3u")
playlist_silence = cue_cut(id="cue_playlist_silence", playlist_silence)
playlist_silence = drop_metadata(playlist_silence)

# Standard Playlists
radio = random(id="standard_playlists", weights=[5], [playlist_silence])

requests = request.queue(id="requests")
requests = cue_cut(id="cue_requests", requests)
radio = fallback(id="requests_fallback", track_sensitive = true, [requests, radio])

interrupting_queue = request.queue(id="interrupting_requests")
interrupting_queue = cue_cut(id="cue_interrupting_requests", interrupting_queue)
radio = fallback(id="interrupting_fallback", track_sensitive = false, [interrupting_queue, radio])

def add_skip_command(s) =
    def skip(_) =
        source.skip(s)
        "Done!"
    end

    server.register(namespace="radio", usage="skip", description="Skip the current song.", "skip",skip)
end

add_skip_command(radio)

# Custom Configuration (Specified in Station Profile)
### BEGIN: BFF.fm Hardcoded Sources
## After: `add_skip_command(radio)`

default_error_source = fallback(track_sensitive = false, [single(!mp3_technical_difficulties), single("/usr/local/share/icecast/web/error.mp3")])

### END: BFF.fm Hardcoded Sources

def live_aware_crossfade(old, new) =
    if !to_live then
        # If going to the live show, play a simple sequence
        sequence([fade.out(old.source),fade.in(new.source)])
    else
        # Otherwise, use the smart transition
        cross.simple(old.source, new.source, fade_in=2.00, fade_out=2.00)
    end
end

radio = cross(minimum=0., duration=3.00, live_aware_crossfade, radio)

# Allow for Telnet-driven insertion of custom metadata.
radio = server.insert_metadata(id="custom_metadata", radio)

# Apply amplification metadata (if supplied)
radio = amplify(override="liq_amplify", 1., radio)

# Normalization and Compression
radio = normalize(target = 0., window = 0.03, gain_min = -16., gain_max = 0., radio)
radio = compress.exponential(radio, mu = 1.0)

radio = fallback(id="safe_fallback", track_sensitive = false, [radio, single(id="error_jingle", "/usr/local/share/icecast/web/error.mp3")])

# Send metadata changes back to AzuraCast
last_title = ref("")
last_artist = ref("")

def metadata_updated(m) =
    def f() =
        if (m["title"] != !last_title or m["artist"] != !last_artist) then
            last_title := m["title"]
            last_artist := m["artist"]

            j = json()

            if (m["song_id"] != "") then
                j.add("song_id", m["song_id"])
                j.add("media_id", m["media_id"])
                j.add("playlist_id", m["playlist_id"])
            else
                j.add("artist", m["artist"])
                j.add("title", m["title"])
            end

            _ = azuracast_api_call(
                "feedback",
                json.stringify(j)
            )
        end
    end

    thread.run(f)
end

radio.on_metadata(metadata_updated)

# Handle "Jingle Mode" tracks by replaying the previous metadata.
last_metadata = ref([])
def handle_jingle_mode(m) =
    if (m["jingle_mode"] == "true") then
        !last_metadata
    else
        last_metadata := m
        m
    end
end

radio = metadata.map(update=false, strip=true, handle_jingle_mode, radio)

# Custom Configuration (Specified in Station Profile)
### BEGIN: BFF.fm Ingest Harbors ###
## After: Azuracast autoDJ setup, `radio = metadata.map(...)`

# A Pre-DJ source of radio powered by the Azuracast automation system
# that can be broadcast as fallback when needed.
#
# Since we've had issues with volume matching and falling back when clients
# connections are unreliable, we just broadcast a silent at the present time.
# At some point, we'll properly figure out how to work with Liquidsoap scheduling
# and/or silence detection to fallback gracefully.
radio_without_live = radio
ignore(radio_without_live)

# To bypass Azuracast AutoDJ, reassign 'radio' here:
#
# ignore(radio)
# If running an automation playlist through AzuraCast while enabling AzuraCast
# live streamer mount, assign `radio_without_live`:
#
# radio = fallback(track_sensitive = false, [radio_without_live, default_error_source])
#
# Otherwise, assign a new error MP3:
#
# radio = default_error_source

# Manage Icecast input harbors remote DJs and Studios

let ingest = ()
let ingest.icecastPort = icecast_port

# Ingest precedence stacks all created HTTP harbors (by name) to create the
# cascade by which on-air is determined:
let ingest.precedence = ref([])
# Active harbor connections and associated user:
let ingest.connections = ref([])

# Connection debugging
# If during development/deployment you need to see the current state of the
# active connections list, call this function.
def ingest.debugConnections() =
  def listToString(str, item) =
    str ^ "#{fst(item)} => #{snd(item)}"
  end
  connectionsDebug = list.fold(listToString, "", !ingest.connections)
  log.debug("Connections: #{connectionsDebug}")
  connectionsDebug
end

def ingest.registerSource(sourceName, highPriority=false) =
  if (highPriority) then
    ingest.precedence := list.add(sourceName, !ingest.precedence)
  else
    ingest.precedence := list.append(!ingest.precedence, [sourceName])
  end
end

# Is a named studio currently connected?
def ingest.isConnected(studio) =
  dj = list.assoc(default="", studio, !ingest.connections)
  (dj != "")
end

# Who is connected to a named studio?
def ingest.djFor(studio) =
  list.assoc(default="", studio, !ingest.connections)
end

def ingest.currentSource() =
  # Get the highest precedence source currently connected
  def getSource(current, source) =
    if (current != "") then
      current
    else
      next = list.assoc(default="", source, !ingest.connections)
      if (next != "") then
        source
      else
        ""
      end
    end
  end
  list.fold(getSource, "", !ingest.precedence)
end

# Who is the current on-air DJ, based on studio precedence?
def ingest.currentDj() =
  # Get the DJ for the highest precedence studio that has a DJ
  def getDj(current, studio) =
    if (current != "") then
      current
    else
      ingest.djFor(studio)
    end
  end
  list.fold(getDj, "", !ingest.precedence)
end

# Global “on connection” handler
def ingest.onDjConnect() =
  source_name = ingest.currentSource()
  dj_name = ingest.currentDj()

  if (dj_name != "") then
    _ = azuracast_api_call(
      timeout_ms=5000,
      "djon",
      json.stringify({user = dj_name})
    )
    slack.log(":rotating_light: ON AIR: #{source_name}/*#{dj_name}*")
  end
end

# Global “on disconnect handler
def ingest.onDjDisconnect() =
  dj_name = ingest.currentDj()
  log.debug("Source disconnected. Current broadcaster was: #{dj_name}")

  if (dj_name != "") then
    _ = azuracast_api_call(
      timeout_ms=5000,
      "djoff",
      json.stringify({user = dj_name})
    )
  end
end

# Handle parsing of username:password strings from legacy clients
def ingest.parseCredentials(login) =
  if (login.user == "source" or login.user == "") and (string.match(pattern="(:|,)+", login.password)) then
    auth_string = string.split(separator="(:|,)", login.password)
    {user = list.nth(default="", auth_string, 0),
    password = list.nth(default="", auth_string, 2)}
  else
    {user = login.user, password = login.password}
  end
end

# Create a studio source
# — These sources have a single fixed password baked into the configuration.
def ingest.makeStudioSource(studioName, ~mount, ~username="source", ~password, ~dropMetadata=false, ~alertOnDisconnect=false, ~highPriority=false) =

  # Register source
  ingest.registerSource(studioName, highPriority)

  def studioAuth(login) =
    auth_info = ingest.parseCredentials(login)

    if ((auth_info.user == username) and (auth_info.password == password )) then
      ingest.connections := list.add((studioName, username), !ingest.connections)
      log.info("Studio connected: #{username}")
      true
    else
      ingest.connections := list.assoc.remove.all(studioName, !ingest.connections)
      log.important("Studio authentication failed: #{username}")
      false
    end
  end

  # When a studio connects, announce to monitoring destinations
  def connected(header) =
    log.info("#{studioName} source connected! - #{header}")
    slack.log(":zap: #{studioName} source connected (*#{ingest.djFor(studioName)}*)")

    # Trigger global connection handler, which will update on-air metadata if
    # this studio is top of the precedence stack
    ingest.onDjConnect()
  end

  # When a studio disconnects, announce to monitoring destinations
  def disconnected() =
    log("#{studioName} disconnected!")

    slack.log(":end: #{studioName} disconnected (*#{ingest.djFor(studioName)}*)")

    # AlertOnDisconnect sends an extra message to our #alerts channel
    if (alertOnDisconnect) then
      slack.log(channel="alerts", ":bangbang: #{studioName} streamer disconnected (*#{ingest.djFor(studioName)}*)")
    end

    ingest.onDjDisconnect()

    # Clear studio DJ variables
    ingest.connections := list.assoc.remove.all(studioName, !ingest.connections)

    # Reset metadata for other remaining connected streams
    ingest.onDjConnect()
  end

  # Create an IceCast harbor for this studio
  harbor = audio_to_stereo(input.harbor(mount, port = ingest.icecastPort, auth = studioAuth, icy = true, icy_metadata_charset = "UTF-8", metadata_charset = "UTF-8", on_connect = connected, on_disconnect = disconnected, buffer = 10., max = 15.))
  ignore(output.dummy(harbor, fallible=true))

  # Experiment: Monitor for connection instability/buffer emptiness
  def interruption_notice() =
    ignore(slack.log(channel="debug", ":leaves: #{studioName} source may be unstable or buffer empty (*#{ingest.djFor(studioName)}*)"))
  end
  harbor.on_leave(interruption_notice)

  # Experiment with silence detection
  def blank_warning() =
    ignore(slack.log(channel="debug", ":mute: #{studioName} Silence detected on source (*#{ingest.djFor(studioName)}*)"))
  end
  ignore(blank.detect(
    max_blank = 5.,
    track_sensitive = false,
    blank_warning,
    harbor
  ))

  # return the source
  if (dropMetadata == true) then
    drop_metadata(harbor)
  else
    harbor
  end
end

# Create an remote DJ source authenticated against AzuraCast's 'streamers' database
def ingest.makeAzuraCastStreamerSource(studioName, ~mount, ~dropMetadata=false, ~highPriority=false) =

  # Register source
  ingest.registerSource(studioName, highPriority)

  # Keep track of who the last DJ was so we can monitor when it changes.
  lastDj = ref("")

  def azuracastAuth(login) =
    auth_info = ingest.parseCredentials(login)
    response = azuracast_api_call(
      timeout_ms=5000,
      "auth",
      json.stringify(auth_info)
    )
    authed = bool_of_string(response)

    if (authed) then
      lastDj := auth_info.user
      log.debug("Set lastDj to: #{!lastDj} (#{auth_info.user})")
    else
      log.important("Remote DJ authentication failed: #{auth_info.user} authentication returned false")
    end

    authed
  end

  def remote_connected(header) =
    dj = !lastDj
    log.debug("Adding (#{studioName}, #{dj}) pair to active connections")
    ingest.connections := list.add((studioName, dj), !ingest.connections)

    log.info("#{studioName} connected! DJ: #{dj} - #{header}")

    # Announce source connection
    slack.log(":zap: #{studioName} source connected (*#{ingest.djFor(studioName)}*)")

    # Trigger global connection handler, which will update on-air metadata if
    # this source is top of the precedence stack
    ingest.onDjConnect()
  end

  def remote_disconnected() =
    log.debug("Remote Disconnected")
    dj = ingest.djFor(studioName)

    log("Remote DJ source disconnected! Was: #{dj}")
    slack.log(":end: #{studioName} disconnected (*#{dj}* :micdrop:)")

    ingest.onDjDisconnect()

    # Clear studio DJ variables
    ingest.connections := list.assoc.remove.all(studioName, !ingest.connections)

    # Reset metadata for other remaining connected streams
    ingest.onDjConnect()
  end

  harbor = audio_to_stereo(input.harbor(mount, port = ingest.icecastPort, auth = azuracastAuth, icy = true, icy_metadata_charset = "UTF-8", metadata_charset = "UTF-8", on_connect = remote_connected, on_disconnect = remote_disconnected, buffer = 10., max = 15.))
  ignore(output.dummy(harbor, fallible=true))

  # return the source
  if (dropMetadata == true) then
    drop_metadata(harbor)
  else
    harbor
  end
end

# Create an remote DJ source authenticated against Creek's user database
def ingest.makeCreekStreamerSource(
  studioName,
  ~mount,
  ~dropMetadata=false,
  ~alertOnDisconnect=false,
  ~authEndpoint="auth/stream",
  ~highPriority=false,
  ~productionEndpoint=true
) =

  # Register source
  if (productionEndpoint) then
    ingest.registerSource(studioName, highPriority)
  end

  # Count successful connections within 10s period to catch ffmpeg crashes
  let successCount = ref(0)
  let successRestartThreshold = 10
  let successRestartWindow = 10.

  def reset_success_count() =
    successCount := 0
  end

  # Keep track of who the last DJ was so we can monitor when it changes.
  lastDj = ref("")

  def creekAuth(login) =
    auth_info = ingest.parseCredentials(login)
    log.debug("Authenticating Creek DJ: #{auth_info.user}")

    authResponse = bff.post(authEndpoint, data=[
      ("username", auth_info.user),
      ("password", auth_info.password)
    ])
    log.debug("Creek DJ auth response: #{authResponse}")

    let json.parse (authResult : {
      authenticated: bool,
      username: string,
      display_name: string?,
      show: string?,
      start_time: string?,
      end_time: string?,
      reason: string?
    }) = authResponse

    if (authResult.authenticated) then
      djName = "#{authResult.display_name ?? authResult.username}"
      showName = "#{authResult.show}"
      endTime = "#{authResult.end_time}"

      lastDj := "#{djName}/#{showName} (#{authResult.username})"
      log.debug("Set lastDj to: #{!lastDj}")

      slack.log(channel="stream_auth", ":key: #{studioName} DJ authenticated: *#{djName}*/*#{showName}* (@#{authResult.username})")
      slack.log(channel="stream_auth", "<@#{authResult.username}> :clock10: Remember to disconnect before *#{timestamp.friendly(endTime)}* when #{showName} ends!")
    else
      # Auth failed, get reason and push to Slack:
      slack.log(channel="stream_auth", "<@#{authResult.username}> :octagonal_sign: #{studioName} authentication failed: Creek user @#{authResult.username} not authenticated because: *#{authResult.reason}*")
      log.important("Creek DJ #{authResult.username} authentication failed: #{authResult.reason}")
    end

    authResult.authenticated
  end

  def remote_connected(header) =
    dj = !lastDj

    # Only record login in connection pool if this is a production endpoint
    if (productionEndpoint) then
      log.debug("Adding (#{studioName}, #{dj}) pair to active connections")
      ingest.connections := list.add((studioName, dj), !ingest.connections)
      def listToString(str, item) =
        str ^ "#{fst(item)} => #{snd(item)}"
      end
      connectionsDebug = list.fold(listToString, "", !ingest.connections)
      log.debug("Connections After Add: #{connectionsDebug}")
    end

    # HACK: Count successful connections in a short time window. If a DJ
    # repeatedly, successfully reconnects within a short span, it's an indicator
    # that the ingest server is crashing at the ffmpeg level. To remedy this, we
    # restart the server.

    # Clear the count afer successRestartWindow seconds (e.g. 10.)
    if (!successCount == 0) then
      thread.run(fast=false, delay=successRestartWindow, reset_success_count)
    end

    successCount := !successCount + 1

    # If this is the nth successful connection within the window, presume crashing is occuring,
    # and restart the server.
    if (!successCount > successRestartThreshold) then
      slack.log(":boom: #{dj} was booted from #{studioName} #{successRestartThreshold} times in #{successRestartWindow} seconds.")
      slack.log(":boom: Ingest server is likely crashing. Automatic recovery triggered: Restarting Azurcast...")
      log.info("#{studioName} ingest crashed! Automatic recovery triggering Azuracast restart...")

      azuracast_public.call("backend/restart")
    end

    log.info("#{studioName} connected! DJ: #{dj} - #{header}")

    # Announce source connection
    slack.log(":zap: #{studioName} source connected (*#{dj}*)")

    # Trigger global connection handler, which will update on-air metadata if
    # this source is top of the precedence stack
    if (productionEndpoint) then
      ingest.onDjConnect()
    end
  end

  def remote_disconnected() =
    log.debug("Remote Disconnected")
    dj = !lastDj

    log("Remote DJ source disconnected! Was: #{dj}")
    slack.log(":end: #{studioName} disconnected (*#{dj}* :micdrop:)")

    # AlertOnDisconnect sends an extra message to our #alerts channel
    if (alertOnDisconnect) then
      slack.log(channel="alerts", ":bangbang: #{studioName} streamer disconnected (*#{ingest.djFor(studioName)}*)")
    end

    if (productionEndpoint) then
      ingest.onDjDisconnect()

      # Clear studio DJ variables
      ingest.connections := list.assoc.remove.all(studioName, !ingest.connections)

      # Reset metadata for other remaining connected streams
      ingest.onDjConnect()
    end
  end

  harbor = audio_to_stereo(input.harbor(mount, port = ingest.icecastPort, auth = creekAuth, icy = true, icy_metadata_charset = "UTF-8", metadata_charset = "UTF-8", on_connect = remote_connected, on_disconnect = remote_disconnected, buffer = 10., max = 15.))
  ignore(output.dummy(harbor, fallible=true))

  # return the source
  if (dropMetadata == true) then
    drop_metadata(harbor)
  else
    harbor
  end
end

# Declare current studio and stream source harbors for which track metadata will be sourced
# from the Creek CMS API
dj_sources = fallback(track_sensitive=false, [
  ingest.makeCreekStreamerSource("Remote DJ", mount="/dj", dropMetadata=true),
  ingest.makeAzuraCastStreamerSource("Remote DJ [Azuracast]", mount="/azdj", dropMetadata=true),
  ingest.makeStudioSource("Studio A", mount="/studio", password=!ingest_studio_a_password, alertOnDisconnect=true, dropMetadata=true),
  # Remote fallback is a 'last resort' source that will only be used if all other sources
  # are disconnected. It may be used in Emergency Broadcast situations, but where we don't
  # want to also displace home broadcasters. If Studio A is frequently reconnecting though,
  # Emergency Broadcast is a better choice as Studio A will cut in over Remote Fallback.
  ingest.makeStudioSource("Remote Fallback", mount="/fallback", username="fallback", password=!ingest_fallback_password, alertOnDisconnect=true, dropMetadata=true)
])

# Tracking metadata
#
# Regardless of which harbor our source is connected from, we treat our CMS as the canonical
# and only source for what music is being played. (Hence `dropMetadata=true` on all the
# Icecast ingestion above.)
#
# Here, we poll the CMS for track changes.
dj_sources = insert_metadata(dj_sources)

# For debugging, we store the last unique track name, and the
# timestamp of when it was registered
last_creek_track = ref(("BFF.fm", timestamp.hms()))

def poll_creek_metadata() =
  # We run the polling in an async thread as a lazy way to ensure that we don't
  # break the polling if there's a JSON parse error or some other crash in the
  # request process. Ensures it'll always run again and failures are shrugged off
  def f() =
    last_name = fst(!last_creek_track)
    now = timestamp.hms()

    # Fetch the now playing track from BFF.fm API
    onAirJson = bff.fetch(cacheBust=true, "data/onair/now.json")
    log.debug("Fetched: #{onAirJson}")

    # Just as an aside, the change in JSON parsing from Liquidsoap 1.4
    # to 2.1 absolutely broke my brain for about half a day, but now that
    # I see it, this is really, really nice.
    let json.parse (onAir : {
      artist: string?,
      title: string?,
      album: string?,
      label: string?,
      program: string?,
      presenter: string?,
      url: string?
    }) = onAirJson

    # If there's no current track (merely a show) then the artist/album/title/label
    # data will be null.
    if (onAir.artist != null()) then
      playingString = "#{onAir.artist} - #{onAir.title}"

      # If the currently playing track isn't the same as on last poll:
      if (playingString != last_name) then
        # Update last displayed track title
        last_creek_track := (playingString, now)

        # Chaged metadata so update source metadata as 'new track'
        dj_sources.insert_metadata(new_track=true, [("title", playingString)])

        # Scrobble new tracks to Last.FM
        lastfm.scrobble(artist="#{onAir.artist}", album="#{onAir.album}", track="#{onAir.title}")

        # Log Now Playing to TuneIn AIR API
        tunein.nowPlaying(artist="#{onAir.artist}", track="#{onAir.title}", album="#{onAir.album ?? onAir.title}")

        # Log to Slack
        slack.log(channel="now_playing", ":notes: #{onAir.artist} - #{onAir.title} (#{onAir.album}) [#{onAir.label ?? ':warning: No Label'}]")
      end
    else
      # No track: Display the current show/host/station

      # Generally, BFF.fm should always have a show in the schedule (even placeholders), but
      # just in case there's a gap, or a data error, set up a fallback
      show = if (onAir.program != null()) then
        "#{onAir.program}"
      else
        "BFF.fm"
      end

      if (show != last_name) then
        # Update last displayed track title
        last_creek_track := (show, now)

        # Update source metadata, but not as a 'track'
        dj_sources.insert_metadata(new_track=false, [("title", show)])

        # Log to Slack
        slack.log(channel="now_playing", ":radio: #{show} w/ #{onAir.presenter ?? 'Unknown Host'}")

        # Log Now Playing 'commerical'/non-music to TuneIn
        tunein.nonMusic(attribution=(onAir.presenter ?? "BFF.fm"), name=show)
      end
    end
  end

  # Run the metadata fetch
  thread.run(f)

  # Return 15s timeout for the next poll
  (15.)
end

# Kick off polling for initial Creek CMS track data
thread.run.recurrent(fast=false, delay=5., poll_creek_metadata)

## Emergency Broadcast
# In addition to the stack of ingest harbors we created above, we create one more mount
# point for “emergency broadcast” — it's set as high priority and will override all other
# Icecast harbors in the stack. (highPriority=true puts it at the top of the harbor stack.)
#
# This is for use in the event that power to a studio goes down, or something goes wrong
# on the main stream, or if aliens invade and we need to organize the resistance from a
# back-up location. That sort of thing.
emergency_broadcast = ingest.makeStudioSource("Emergency Broadcast", mount="/emergency", username="ebs", password=!ingest_emergency_password, highPriority=true)

live_broadcast = fallback(track_sensitive=false, [emergency_broadcast, dj_sources])

# From the Azuracast harbor config, add 'is_live' metadata
# Unlike the Azuracast default, we don't add default text.
def insert_live_metadata(m) =
  ignore(m)
  [("is_live", "true")]
end
live_broadcast = map_metadata(insert_live_metadata, live_broadcast)

## Reapply crossfade to new sources.
# TODO: This works, but is glitchy. Need to dedicate some time to properly
# learning Liquidsoap crossfade construction, especially when we move automation
# into the cloud.
# DEBUGGING: Remove call to crossfade as it appears to get triggered whenever track metadata changes
# live_broadcast = cross(minimum=0., duration=3.00, live_aware_crossfade, live_broadcast)

radio = fallback(track_sensitive=false, [live_broadcast, radio])

# From the default Azuracast harbor behavior, trigger automation track skipping
# when going live.
def check_live() =
  if live_broadcast.is_ready() then
    if not !to_live then
      to_live := true
      radio_without_live.skip()
    end
  else
    to_live := false
  end
end
radio = source.on_frame(radio, check_live)

# Trigger metadata updates back to Azuracast:
radio.on_metadata(metadata_updated)

## Create Creek Authentication test Harbor
# This dedicated harbor (and output) provides a channel for DJs to activate an
# icecast client and verify their user/pass credentials are being handled correctly.
#
# The production auth endpoint in Creek enforces the user's show schedule time, whereas
# the /test mount can be used any time.
test_dj = ingest.makeCreekStreamerSource("Creek DJ Auth Tester", mount="/test", dropMetadata=true, authEndpoint="auth/stream/test", productionEndpoint=false)
test_dj = fallback(track_sensitive=false, [test_dj, single(!mp3_test_feed), default_error_source])
ignore(output.dummy(test_dj, fallible=true))
output.icecast(%mp3(samplerate=44100, stereo=true, bitrate=128, id3v2=true),
  host = "127.0.0.1", port = 8000, password = !icecast_password,
  mount = "/test.mp3",
  name = "BFF.fm Broadcast System Test Channel",
  description = "For DJs to test their Icecast configurations",
  genre = "Unclassifiable",
  url = !user_agent_url,
  public = false,
  encoding = "UTF-8",
  test_dj
)

### END: BFF.fm Ingest Harbors ###

# Local Broadcasts
output.icecast(%mp3(samplerate=44100, stereo=true, bitrate=128, id3v2=true), id="local_1", host = "127.0.0.1", port = 8000, password = "(PASSWORD)", mount = "/radio.mp3", name = "BFF.fm Station Upgrade Testbed", description = "Liquidsoap 1.4 - 2.1 upgrade", genre = "Unclassifiable", url = "https://bff.fm", public = false, encoding = "UTF-8", radio)

# Remote Relays

# Custom Configuration (Specified in Station Profile)
### BEGIN: BFF.fm Startup/Shutdown Notices
## After: Remote Relays

def start_notices() =
  log("Liquidsoap started.")
  slack.log(":small_red_triangle: Liquidsoap started.")
end

on_start(start_notices)

def shutdown_notices() =
  log("Liquidsoap shut down.")
  slack.log(":small_red_triangle_down: Liquidsoap shutdown.")
end

on_shutdown(shutdown_notices)
### END: BFF.fm Startup/Shutdown Notices