tribes-plugin-sender/lib/sender/streaming/stream_generation.ex

defmodule TribeOne.TribesPlugin.Sender.Streaming.StreamGeneration do
  @moduledoc false

  use Ash.Resource,
    otp_app: :tribe_one_sender,
    domain: TribeOne.TribesPlugin.Sender.Streaming,
    data_layer: AshPostgres.DataLayer,
    authorizers: [Ash.Policy.Authorizer],
    extensions: [AshNostrSync]

  @statuses [:pending, :listening, :live, :degraded, :ending, :ended, :failed]
  @delivery_modes [:hls, :ll_hls]

  postgres do
    table("sender_stream_generations")
    repo(Tribes.Repo)

    custom_indexes do
      index([:stream_id])
      index([:status])
      index([:active_ingest_endpoint_id])
      index([:primary_origin_endpoint_id])
    end
  end

  nostr_sync do
    namespace("plugins.sender.stream_generation")
    lane(:control)
    publish?(true)
    consume?(true)
  end

  actions do
    defaults([:read])

    read :by_id do
      get?(true)

      argument :id, :uuid do
        allow_nil?(false)
      end

      filter(expr(id == ^arg(:id)))
    end

    create :create do
      accept([
        :stream_id,
        :status,
        :started_at,
        :ended_at,
        :active_ingest_endpoint_id,
        :primary_origin_endpoint_id,
        :delivery_mode,
        :generation_token,
        :playlist_namespace,
        :error
      ])

      change(AshNostrSync.PublishChange)
    end

    create :sync_upsert do
      accept([
        :id,
        :stream_id,
        :status,
        :started_at,
        :ended_at,
        :active_ingest_endpoint_id,
        :primary_origin_endpoint_id,
        :delivery_mode,
        :generation_token,
        :playlist_namespace,
        :error
      ])

      upsert?(true)
    end

    update :update do
      require_atomic?(false)

      accept([
        :status,
        :started_at,
        :ended_at,
        :active_ingest_endpoint_id,
        :primary_origin_endpoint_id,
        :delivery_mode,
        :generation_token,
        :playlist_namespace,
        :error
      ])

      change(AshNostrSync.PublishChange)
    end
  end

  policies do
    bypass Tribes.Checks.SyncInteraction do
      authorize_if(always())
    end

    bypass Tribes.Checks.SystemInteraction do
      authorize_if(always())
    end

    policy action_type(:read) do
      authorize_if(always())
    end
  end

  attributes do
    attribute :id, :uuid do
      allow_nil?(false)
      primary_key?(true)
      public?(true)
      writable?(true)
      default(&Ash.UUID.generate/0)
    end

    attribute :stream_id, :uuid do
      allow_nil?(false)
      public?(true)
    end

    attribute :status, :atom do
      constraints(one_of: @statuses)
      allow_nil?(false)
      default(:pending)
      public?(true)
    end

    attribute :started_at, :utc_datetime do
      public?(true)
    end

    attribute :ended_at, :utc_datetime do
      public?(true)
    end

    attribute :active_ingest_endpoint_id, :uuid do
      public?(true)
    end

    attribute :primary_origin_endpoint_id, :uuid do
      public?(true)
    end

    attribute :delivery_mode, :atom do
      constraints(one_of: @delivery_modes)
      allow_nil?(false)
      default(:hls)
      public?(true)
    end

    attribute :generation_token, :string do
      public?(true)
    end

    attribute :playlist_namespace, :string do
      public?(true)
    end

    attribute :error, :string do
      public?(true)
    end

    timestamps(type: :utc_datetime)
  end
end