Create feed segments (#494)

* Create FeedSegment, update listing timestamps with callback function

* Update feed segment creation to use bulk insertion

* Fix typo

* Add feed_stream_id filter for querying feed segments index

* Add json apis for feed_streams and feed_segments

* Add unique worker for updating feed segments

* Fix typo with selecting continuation token

server/config/config.exs

@@ -93,7 +93,7 @@ config :orcasite, Oban,
   repo: Orcasite.Repo,
   # 7 day job retention
   plugins: [{Oban.Plugins.Pruner, max_age: 7 * 24 * 60 * 60}],
-  queues: [default: 10, email: 10]
+  queues: [default: 10, email: 10, feed_segments: 10]
 
 config :spark, :formatter,
   remove_parens?: true,

server/lib/orcasite/application.ex

@@ -23,7 +23,7 @@ defmodule Orcasite.Application do
         {Finch, name: Orcasite.Finch},
         {Task.Supervisor, name: Orcasite.TaskSupervisor},
         {AshAuthentication.Supervisor, otp_app: :orcasite},
-        if(Application.get_env(:orcasite, :feed_stream_queue_url),
+        if(Application.get_env(:orcasite, :feed_stream_queue_url) not in [nil, ""],
           do: {Orcasite.Radio.FeedStreamQueue, []}
         ),
         OrcasiteWeb.Endpoint

server/lib/orcasite/global_setup.ex

@@ -4,11 +4,12 @@ defmodule Orcasite.GlobalSetup do
     |> Ash.Query.for_read(:read)
     |> Orcasite.Radio.read!()
     |> Stream.map(fn feed ->
-      with {:ok, %{timestamps: timestamps}} <- Orcasite.Radio.AwsClient.list_timestamps(feed) do
+      Orcasite.Radio.AwsClient.list_timestamps(feed, fn timestamps ->
         timestamps
         |> Enum.map(&%{feed: feed, playlist_timestamp: &1})
         |> Orcasite.Radio.bulk_create(Orcasite.Radio.FeedStream, :create)
-      end
+      end)
+      :ok
     end)
     |> Enum.to_list()
   end

server/lib/orcasite/radio/aws_client.ex

@@ -1,15 +1,30 @@
 defmodule Orcasite.Radio.AwsClient do
-  alias Orcasite.Radio.Feed
+  alias Orcasite.Radio.{Feed, FeedStream}
 
   @default_results %{count: 0, timestamps: []}
 
-  def list_timestamps(%Feed{} = feed) do
-    loop_request_timestamp(feed)
+  def get_feed_stream(%FeedStream{
+        bucket_region: bucket_region,
+        bucket: bucket,
+        playlist_m3u8_path: path
+      }) do
+    ExAws.S3.get_object(bucket, path)
+    |> ExAws.request(region: bucket_region)
+    |> case do
+      {:ok, %{body: body, status_code: 200}} -> {:ok, body}
+      {:ok, other} -> {:error, other}
+      {:error, error} -> {:error, error}
+    end
+  end
+
+  def list_timestamps(%Feed{} = feed, callback \\ nil) do
+    loop_request_timestamp(feed, callback)
   end
 
   def request_timestamps(
         %Feed{node_name: node_name, bucket: bucket, bucket_region: bucket_region},
-        continuation_token \\ nil
+        continuation_token \\ nil,
+        callback \\ nil
       ) do
     continuation =
       case continuation_token do
@@ -18,7 +33,7 @@ defmodule Orcasite.Radio.AwsClient do
         token -> [continuation_token: token]
       end
 
-    options = [prefix: "#{node_name}/hls/", delimiter: "/", max_keys: 1000] ++ continuation
+    options = [prefix: "#{node_name}/hls/", delimiter: "/", max_keys: 100] ++ continuation
 
     ExAws.S3.list_objects_v2(bucket, options)
     |> ExAws.request(region: bucket_region)
@@ -33,9 +48,13 @@ defmodule Orcasite.Radio.AwsClient do
 
         count = Enum.count(timestamps)
 
-        token = body |> Map.get(:continuation_token)
+        token = body |> Map.get(:next_continuation_token)
         has_more = body |> Map.get(:is_truncated) == "true"
 
+        if is_function(callback) do
+          callback.(timestamps)
+        end
+
         {:ok,
          %{count: count, timestamps: timestamps, continuation_token: token, has_more: has_more}}
 
@@ -44,8 +63,8 @@ defmodule Orcasite.Radio.AwsClient do
     end
   end
 
-  def loop_request_timestamp(feed, token \\ nil, results \\ @default_results) do
-    request_timestamps(feed, token)
+  def loop_request_timestamp(feed, callback \\ nil, token \\ nil, results \\ @default_results) do
+    request_timestamps(feed, token, callback)
     |> case do
       {:ok,
        %{
@@ -54,7 +73,7 @@ defmodule Orcasite.Radio.AwsClient do
          timestamps: timestamps,
          continuation_token: continuation_token
        }} ->
-        loop_request_timestamp(feed, continuation_token, %{
+        loop_request_timestamp(feed, callback, continuation_token, %{
           count: count + results.count,
           timestamps: results.timestamps ++ timestamps
         })

server/lib/orcasite/radio/feed_segment.ex

@@ -0,0 +1,181 @@
+defmodule Orcasite.Radio.FeedSegment do
+  use Ash.Resource,
+    extensions: [AshAdmin.Resource, AshUUID, AshGraphql.Resource, AshJsonApi.Resource],
+    data_layer: AshPostgres.DataLayer
+
+  postgres do
+    table "feed_segments"
+    repo Orcasite.Repo
+
+    migration_defaults id: "fragment(\"uuid_generate_v7()\")"
+
+    custom_indexes do
+      index [:start_time]
+      index [:end_time]
+      index [:feed_id]
+      index [:feed_stream_id]
+      index [:bucket]
+    end
+  end
+
+  identities do
+    identity :feed_segment_timestamp, [:feed_id, :start_time]
+    identity :feed_segment_path, [:segment_path]
+  end
+
+  attributes do
+    uuid_attribute :id, prefix: "fdseg"
+
+    attribute :start_time, :utc_datetime
+    attribute :end_time, :utc_datetime
+    attribute :duration, :decimal
+
+    attribute :bucket, :string
+    attribute :bucket_region, :string
+    attribute :cloudfront_url, :string
+
+    attribute :playlist_timestamp, :string do
+      description "UTC Unix epoch for playlist (m3u8 dir) start (e.g. 1541027406)"
+    end
+
+    attribute :playlist_path, :string do
+      description "S3 object path for playlist dir (e.g. /rpi_orcasound_lab/hls/1541027406/)"
+    end
+
+    attribute :playlist_m3u8_path, :string do
+      description "S3 object path for playlist file (e.g. /rpi_orcasound_lab/hls/1541027406/live.m3u8)"
+    end
+
+    attribute :segment_path, :string do
+      description "S3 object path for ts file (e.g. /rpi_orcasound_lab/hls/1541027406/live005.ts)"
+    end
+
+    attribute :file_name, :string do
+      description "ts file name (e.g. live005.ts)"
+      allow_nil? false
+    end
+
+    create_timestamp :inserted_at
+    update_timestamp :updated_at
+  end
+
+  relationships do
+    belongs_to :feed, Orcasite.Radio.Feed
+    belongs_to :feed_stream, Orcasite.Radio.FeedStream
+  end
+
+  actions do
+    defaults [:read, :update, :destroy]
+
+    read :index do
+      pagination do
+        offset? true
+        countable true
+        default_limit 100
+      end
+
+      argument :feed_id, :string
+      argument :feed_stream_id, :string
+
+      filter expr(
+               if(not is_nil(^arg(:feed_id)), do: feed_id == ^arg(:feed_id), else: true) and
+                 if(
+                   not is_nil(^arg(:feed_stream_id)),
+                   do: feed_stream_id == ^arg(:feed_stream_id),
+                   else: true
+                 )
+             )
+    end
+
+    create :create do
+      primary? true
+      upsert? true
+      upsert_identity :feed_segment_path
+
+      upsert_fields [
+        :playlist_path,
+        :duration,
+        :start_time,
+        :end_time,
+        :bucket,
+        :bucket_region,
+        :cloudfront_url,
+        :playlist_timestamp,
+        :playlist_m3u8_path,
+        :file_name
+      ]
+
+      accept [
+        :start_time,
+        :end_time,
+        :duration,
+        :bucket,
+        :bucket_region,
+        :cloudfront_url,
+        :playlist_timestamp,
+        :playlist_m3u8_path,
+        :playlist_path,
+        :file_name
+      ]
+
+      argument :feed, :map, allow_nil?: false
+      argument :feed_stream, :map, allow_nil?: false
+
+      argument :segment_path, :string do
+        allow_nil? false
+        constraints allow_empty?: false
+      end
+
+      change set_attribute(:segment_path, arg(:segment_path))
+      change manage_relationship(:feed, type: :append)
+      change manage_relationship(:feed_stream, type: :append)
+
+      change fn changeset, context ->
+        feed = Ash.Changeset.get_argument_or_attribute(changeset, :feed)
+
+        playlist_timestamp =
+          changeset
+          |> Ash.Changeset.get_attribute(:playlist_timestamp)
+
+        feed
+        |> Map.take([:bucket, :bucket_region, :cloudfront_url])
+        |> Enum.reduce(changeset, fn {attribute, value}, acc ->
+          acc
+          |> Ash.Changeset.change_new_attribute(attribute, value)
+        end)
+        |> Ash.Changeset.change_new_attribute(
+          :start_time,
+          playlist_timestamp
+          |> String.to_integer()
+          |> DateTime.from_unix!()
+        )
+        |> Ash.Changeset.change_new_attribute(
+          :playlist_path,
+          "/#{feed.node_name}/hls/#{playlist_timestamp}/"
+        )
+        |> Ash.Changeset.change_new_attribute(
+          :playlist_m3u8_path,
+          "/#{feed.node_name}/hls/#{playlist_timestamp}/live.m3u8"
+        )
+      end
+    end
+  end
+
+  json_api do
+    type "feed_segment"
+
+    routes do
+      base "/feed_segments"
+
+      index :index
+    end
+  end
+
+  graphql do
+    type :feed_segment
+
+    queries do
+      list :feed_segments, :index
+    end
+  end
+end