Stream recording UI

.prettierignore

@@ -2,6 +2,7 @@
 app/assets/builds
 build
 coverage
+tmp
 
 # Ignore vendor'd libraries
 vendor

app/controllers/podcast_stream_controller.rb

@@ -0,0 +1,47 @@
+class PodcastStreamController < ApplicationController
+  before_action :set_stream
+
+  def show
+  end
+
+  def update
+    @stream.assign_attributes(stream_params)
+
+    respond_to do |format|
+      if @stream.save
+        format.html { redirect_to podcast_stream_path(@podcast), notice: t(".notice") }
+      else
+        flash.now[:error] = t(".error")
+        format.html { render :show, status: :unprocessable_entity }
+      end
+    end
+  rescue ActiveRecord::StaleObjectError
+    render :show, status: :conflict
+  end
+
+  private
+
+  def set_stream
+    @podcast = Podcast.find(params[:podcast_id])
+    @stream = @podcast.stream_recording || @podcast.build_stream_recording
+    @stream.clear_attribute_changes(%i[podcast_id])
+    @stream.locking_enabled = true
+    authorize @stream
+  rescue ActiveRecord::RecordNotFound => e
+    render_not_found(e)
+  end
+
+  def stream_params
+    nilify params.fetch(:stream_recording, {}).permit(
+      :lock_version,
+      :url,
+      :status,
+      :start_date,
+      :end_date,
+      :create_as,
+      :expiration,
+      record_days: [],
+      record_hours: []
+    )
+  end
+end

app/helpers/stream_recordings_helper.rb

@@ -0,0 +1,27 @@
+module StreamRecordingsHelper
+  def stream_status_options
+    StreamRecording.statuses.keys.map { |k| [I18n.t("helpers.label.stream_recording.statuses.#{k}"), k] }
+  end
+
+  def stream_create_as_options
+    StreamRecording.create_as.keys.map { |k| [I18n.t("helpers.label.stream_recording.create_as_opts.#{k}"), k] }
+  end
+
+  def stream_expiration_options
+    I18n.t("helpers.label.stream_recording.expirations").invert.to_a
+  end
+
+  def stream_record_days_options(val)
+    label = I18n.t("helpers.label.stream_recording.record_all_days")
+    all = [label, "", {selected: val.blank?, data: {mandatory: true}}]
+    opts = StreamRecording::ALL_DAYS.map { |d| [I18n.t("date.day_names")[d % 7], d] }
+    options_for_select(opts.prepend(all), val)
+  end
+
+  def stream_record_hours_options(val)
+    label = I18n.t("helpers.label.stream_recording.record_all_hours")
+    all = [label, "", {selected: val.blank?, data: {mandatory: true}}]
+    opts = StreamRecording::ALL_HOURS.map { |h| [Time.new(2000, 1, 1, h, 0, 0).strftime("%l %p").strip, h] }
+    options_for_select(opts.prepend(all), val)
+  end
+end

app/javascript/controllers/slim_select_controller.js

@@ -2,49 +2,18 @@ import { Controller } from "@hotwired/stimulus"
 import SlimSelect from "slim-select"
 
 export default class extends Controller {
-  static values = { groupSelect: Boolean, exclusive: Array }
+  static values = { exclusive: Array }
 
   connect() {
     this.select = new SlimSelect({
       select: this.element,
       settings: {
-        // TODO: broken
-        // selectByGroup: this.groupSelectValue,
         placeholderText: "",
         allowDeselect: this.hasEmpty(this.element),
         showSearch: this.showSearch(this.element),
       },
       events: {
-        afterChange: (val) => {
-          if (val.length > 0) {
-            this.select.selectEl.classList.remove("form-control-blank")
-          } else {
-            this.select.selectEl.classList.add("form-control-blank")
-          }
-          this.element.dispatchEvent(new Event("blur"))
-        },
-        beforeChange: (newOpts, oldOpts) => {
-          if (this.exclusiveValue.length) {
-            const newVals = newOpts.map((o) => o.value)
-            const oldVals = oldOpts.map((o) => o.value)
-            const added = newVals.find((v) => !oldVals.includes(v))
-            const addedExclusive = this.exclusiveValue.includes(added) ? added : null
-            const addedNonExclusive = !this.exclusiveValue.includes(added) ? added : null
-
-            // deselect when adding an exclusive
-            if (addedExclusive && oldVals.length) {
-              this.select.setSelected([addedExclusive])
-              return false
-            }
-
-            // deselect if exclusive is selected and we added non-exclusive
-            if (oldVals.find((v) => this.exclusiveValue.includes(v)) && addedNonExclusive) {
-              this.select.setSelected([addedNonExclusive])
-              return false
-            }
-          }
-          return true
-        },
+        beforeChange: this.beforeChange.bind(this),
       },
     })
   }
@@ -66,4 +35,45 @@ export default class extends Controller {
   showSearch(element) {
     return element.children.length > 10
   }
+
+  beforeChange(newVal, oldVal) {
+    if (this.exclusiveValue.length) {
+      const newVals = newVal.map((o) => o.value)
+      const oldVals = oldVal.map((o) => o.value)
+      const added = newVals.find((v) => !oldVals.includes(v))
+      const addedExclusive = this.exclusiveValue.includes(added) ? added : null
+      const addedNonExclusive = !this.exclusiveValue.includes(added) ? added : null
+
+      // deselect when adding an exclusive
+      if (addedExclusive !== null && oldVals.length) {
+        this.selectLater([addedExclusive])
+      }
+
+      // deselect if exclusive is selected and we added non-exclusive
+      if (oldVals.some((v) => this.exclusiveValue.includes(v)) && addedNonExclusive !== null) {
+        this.selectLater([addedNonExclusive])
+      }
+    }
+
+    // fix material-style floating labels as selection changes
+    this.blankLater(newVal)
+
+    return true
+  }
+
+  selectLater(val) {
+    setTimeout(() => {
+      this.select.setSelected([val])
+    }, 1)
+  }
+
+  blankLater(val) {
+    setTimeout(() => {
+      if (val.length > 0) {
+        this.select.selectEl.classList.remove("form-control-blank")
+      } else {
+        this.select.selectEl.classList.add("form-control-blank")
+      }
+    }, 1)
+  }
 }

app/jobs/stream_recording_config_job.rb

@@ -0,0 +1,15 @@
+class StreamRecordingConfigJob < ApplicationJob
+  S3_KEY = "streams.json"
+
+  queue_as :feeder_default
+
+  def perform
+    s3_client.put_object(
+      body: StreamRecording.config.to_json,
+      bucket: s3_bucket,
+      cache_control: "max-age=60",
+      content_type: "application/json",
+      key: S3_KEY
+    )
+  end
+end

app/models/application_record.rb

@@ -29,4 +29,11 @@ def locking_enabled?
   def stale?
     !!try(:lock_version_changed?)
   end
+
+  def set_default(key, value)
+    if has_attribute?(key)
+      self[key] ||= value
+      clear_attribute_changes([key]) if new_record? && self[key] == value
+    end
+  end
 end

app/models/concerns/porter_utils.rb

@@ -15,6 +15,13 @@ def porter_region
     end
   end
 
+  def self.callback_sqs
+    region = ENV["AWS_REGION"].present? ? ENV["AWS_REGION"] : "us-east-1"
+    account = ENV["AWS_ACCOUNT_ID"]
+    queue = ApplicationWorker.prefix_name("fixer_callback")
+    "https://sqs.#{region}.amazonaws.com/#{account}/#{queue}"
+  end
+
   def porter_start!(job)
     self.class.porter_sns_client.publish(topic_arn: ENV["PORTER_SNS_TOPIC"], message: {Job: job}.to_json)
   end
@@ -35,14 +42,10 @@ def porter_tasks
   end
 
   def porter_callbacks
-    region = ENV["AWS_REGION"].present? ? ENV["AWS_REGION"] : "us-east-1"
-    account = ENV["AWS_ACCOUNT_ID"]
-    queue = ApplicationWorker.prefix_name("fixer_callback")
-
     [
       {
         Type: "AWS/SQS",
-        Queue: "https://sqs.#{region}.amazonaws.com/#{account}/#{queue}"
+        Queue: PorterUtils.callback_sqs
       }
     ]
   end

app/models/podcast.rb

@@ -25,6 +25,7 @@ class Podcast < ApplicationRecord
 
   has_one :default_feed, -> { default }, class_name: "Feed", validate: true, autosave: true, inverse_of: :podcast
   alias_method :public_feed, :default_feed
+  has_one :stream_recording, validate: true, autosave: true
 
   has_many :episodes, -> { order("published_at desc") }, dependent: :destroy
   has_many :feeds, dependent: :destroy

app/models/stream_recording.rb

@@ -0,0 +1,70 @@
+class StreamRecording < ApplicationRecord
+  ALL_DAYS = (1..7).to_a
+  ALL_HOURS = (0..23).to_a
+
+  enum :status, %w[enabled disabled paused].to_enum_h, prefix: true
+  enum :create_as, %w[clips episodes].to_enum_h, prefix: true
+
+  serialize :record_days, coder: JSON
+  serialize :record_hours, coder: JSON
+
+  belongs_to :podcast, -> { with_deleted }, touch: true, optional: true
+
+  scope :active, ->(now = Time.now) { status_enabled.where("end_date IS NULL OR end_date > ?", now) }
+  scope :recording, ->(now = Time.now) { active.where("start_date > ?", now) }
+
+  validates :url, presence: true, http_url: true, http_head: /audio\/.+/
+  validates :start_date, presence: true, if: :status_enabled?
+  validates :end_date, comparison: {greater_than: :start_date}, allow_nil: true, if: :start_date
+  validates :record_days, inclusion: {in: ALL_DAYS}, allow_nil: true
+  validates :record_hours, inclusion: {in: ALL_HOURS}, allow_nil: true
+  validates :expiration, numericality: {greater_than: 0}, allow_nil: true
+
+  after_initialize :set_defaults
+  after_save :write_config
+
+  acts_as_paranoid
+
+  def self.config
+    active.map do |s|
+      {
+        id: s.id,
+        gid: s.to_global_id.to_s,
+        podcast_id: s.podcast_id,
+        url: s.url,
+        start_date: s.start_date,
+        end_date: s.end_date,
+        record_days: s.record_days,
+        record_hours: s.record_hours,
+        callback: PorterUtils.callback_sqs
+      }
+    end
+  end
+
+  def set_defaults
+    set_default(:status, "disabled")
+    set_default(:create_as, "clips")
+  end
+
+  def write_config
+    StreamRecordingConfigJob.perform_later
+  end
+
+  def record_days=(val)
+    days = Array(val).reject(&:blank?).map(&:to_i).uniq.sort
+    if days.empty? || days == ALL_DAYS
+      super(nil)
+    else
+      super(days)
+    end
+  end
+
+  def record_hours=(val)
+    hours = Array(val).reject(&:blank?).map(&:to_i).uniq.sort
+    if hours.empty? || hours == ALL_HOURS
+      super(nil)
+    else
+      super(hours)
+    end
+  end
+end

app/models/validators/http_head_validator.rb

@@ -0,0 +1,46 @@
+class HttpHeadValidator < ActiveModel::EachValidator
+  include HttpUtil
+
+  def self.skip_validation?
+    Rails.env.test?
+  end
+
+  def validate_each(rec, attr, val)
+    return if self.class.skip_validation?
+
+    if val.present? && HttpUrlValidator.http_url?(val) && (rec.new_record? || rec.changes[attr].present?)
+      res = memoize_http_head(val)
+
+      if res.nil? || !res.is_a?(Net::HTTPSuccess)
+        rec.errors.add(attr, :unreachable, message: "not http reachable")
+      elsif options[:with] && !has_content_type?(res, options[:with])
+        rec.errors.add(attr, :invalid_content_type, message: "invalid content type")
+      end
+    end
+  end
+
+  private
+
+  def memoize_http_head(url)
+    if @last_http_head == url
+      @last_http_res
+    else
+      @last_http_head = url
+      @last_http_res = http_head(url)
+    end
+  end
+
+  def has_content_type?(res, matching)
+    type = res.content_type.to_s.downcase
+
+    if options[:with].is_a?(String) && options[:with] == type
+      true
+    elsif options[:with].is_a?(Array) && options[:with].include?(type)
+      true
+    elsif options[:with].is_a?(Regexp) && options[:with].match?(type)
+      true
+    else
+      false
+    end
+  end
+end

app/policies/stream_recording_policy.rb

@@ -0,0 +1,13 @@
+class StreamRecordingPolicy < ApplicationPolicy
+  def show?
+    PodcastPolicy.new(token, resource.podcast).show?
+  end
+
+  def create?
+    update?
+  end
+
+  def update?
+    PodcastPolicy.new(token, resource.podcast).update?
+  end
+end