akkoma/lib/pleroma/web/api_spec/operations/streaming_operation.ex

# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only

defmodule Pleroma.Web.ApiSpec.StreamingOperation do
  alias OpenApiSpex.Operation
  alias OpenApiSpex.Response
  alias OpenApiSpex.Schema
  alias Pleroma.Web.ApiSpec.NotificationOperation
  alias Pleroma.Web.ApiSpec.Schemas.Chat
  alias Pleroma.Web.ApiSpec.Schemas.Conversation
  alias Pleroma.Web.ApiSpec.Schemas.FlakeID
  alias Pleroma.Web.ApiSpec.Schemas.Status

  require Pleroma.Constants

  @spec open_api_operation(atom) :: Operation.t()
  def open_api_operation(action) do
    operation = String.to_existing_atom("#{action}_operation")
    apply(__MODULE__, operation, [])
  end

  @spec streaming_operation() :: Operation.t()
  def streaming_operation do
    %Operation{
      tags: ["Timelines"],
      summary: "Establish streaming connection",
      description: "Receive statuses in real-time via WebSocket.",
      security: [%{"oAuth" => ["read:statuses", "read:notifications"]}],
      operationId: "WebsocketHandler.streaming",
      parameters: [
        Operation.parameter(:connection, :header, %Schema{type: :string}, "connection header",
          required: true
        ),
        Operation.parameter(:upgrade, :header, %Schema{type: :string}, "upgrade header",
          required: true
        ),
        Operation.parameter(
          :"sec-websocket-key",
          :header,
          %Schema{type: :string},
          "sec-websocket-key header",
          required: true
        ),
        Operation.parameter(
          :"sec-websocket-version",
          :header,
          %Schema{type: :string},
          "sec-websocket-version header",
          required: true
        )
      ],
      requestBody: request_body("Client-sent events", client_sent_events()),
      responses: %{
        101 => switching_protocols_response(),
        200 =>
          Operation.response(
            "Server-sent events",
            "application/json",
            server_sent_events()
          )
      }
    }
  end

  defp switching_protocols_response do
    %Response{
      description: "Switching protocols",
      headers: %{
        "connection" => %OpenApiSpex.Header{required: true},
        "upgrade" => %OpenApiSpex.Header{required: true},
        "sec-websocket-accept" => %OpenApiSpex.Header{required: true}
      }
    }
  end

  defp server_sent_events do
    %Schema{
      oneOf: [
        update_event(),
        status_update_event(),
        notification_event(),
        chat_update_event(),
        follow_relationships_update_event(),
        conversation_event(),
        delete_event(),
        pleroma_respond_event()
      ]
    }
  end

  defp stream do
    %Schema{
      type: :array,
      title: "Stream",
      description: """
      The stream identifier.
      The first item is the name of the stream. If the stream needs a differentiator, the second item will be the corresponding identifier.
      Currently, for the following stream types, there is a second element in the array:

      - `list`: The second element is the id of the list, as a string.
      - `hashtag`: The second element is the name of the hashtag.
      - `public:remote:media` and `public:remote`: The second element is the domain of the corresponding instance.
      """,
      maxItems: 2,
      minItems: 1,
      items: %Schema{type: :string},
      example: ["hashtag", "mew"]
    }
  end

  defp get_schema(%Schema{} = schema), do: schema
  defp get_schema(schema), do: schema.schema

  defp server_sent_event_helper(name, description, type, payload, opts \\ []) do
    payload_type = Keyword.get(opts, :payload_type, :json)
    has_stream = Keyword.get(opts, :has_stream, true)

    stream_properties =
      if has_stream do
        %{stream: stream()}
      else
        %{}
      end

    stream_example = if has_stream, do: %{"stream" => get_schema(stream()).example}, else: %{}

    stream_required = if has_stream, do: [:stream], else: []

    payload_schema =
      if payload_type == :json do
        %Schema{
          title: "Event payload",
          description: "JSON-encoded string of #{get_schema(payload).title}",
          allOf: [payload]
        }
      else
        payload
      end

    payload_example =
      if payload_type == :json do
        get_schema(payload).example |> Jason.encode!()
      else
        get_schema(payload).example
      end

    %Schema{
      type: :object,
      title: name,
      description: description,
      required: [:event, :payload] ++ stream_required,
      properties:
        %{
          event: %Schema{
            title: "Event type",
            description: "Type of the event.",
            type: :string,
            required: true,
            enum: [type]
          },
          payload: payload_schema
        }
        |> Map.merge(stream_properties),
      example:
        %{
          "event" => type,
          "payload" => payload_example
        }
        |> Map.merge(stream_example)
    }
  end

  defp update_event do
    server_sent_event_helper("New status", "A newly-posted status.", "update", Status)
  end

  defp status_update_event do
    server_sent_event_helper("Edit", "A status that was just edited", "status.update", Status)
  end

  defp notification_event do
    server_sent_event_helper(
      "Notification",
      "A new notification.",
      "notification",
      NotificationOperation.notification()
    )
  end

  defp follow_relationships_update_event do
    server_sent_event_helper(
      "Follow relationships update",
      "An update to follow relationships.",
      "pleroma:follow_relationships_update",
      %Schema{
        type: :object,
        title: "Follow relationships update",
        required: [:state, :follower, :following],
        properties: %{
          state: %Schema{
            type: :string,
            description: "Follow state of the relationship.",
            enum: ["follow_pending", "follow_accept", "follow_reject", "unfollow"]
          },
          follower: %Schema{
            type: :object,
            description: "Information about the follower.",
            required: [:id, :follower_count, :following_count],
            properties: %{
              id: FlakeID,
              follower_count: %Schema{type: :integer},
              following_count: %Schema{type: :integer}
            }
          },
          following: %Schema{
            type: :object,
            description: "Information about the following person.",
            required: [:id, :follower_count, :following_count],
            properties: %{
              id: FlakeID,
              follower_count: %Schema{type: :integer},
              following_count: %Schema{type: :integer}
            }
          }
        },
        example: %{
          "state" => "follow_pending",
          "follower" => %{
            "id" => "someUser1",
            "follower_count" => 1,
            "following_count" => 1
          },
          "following" => %{
            "id" => "someUser2",
            "follower_count" => 1,
            "following_count" => 1
          }
        }
      }
    )
  end

  defp chat_update_event do
    server_sent_event_helper(
      "Chat update",
      "A new chat message.",
      "pleroma:chat_update",
      Chat
    )
  end

  defp conversation_event do
    server_sent_event_helper(
      "Conversation",
      "An update about a conversation",
      "conversation",
      Conversation
    )
  end

  defp delete_event do
    server_sent_event_helper(
      "Delete",
      "A status that was just deleted.",
      "delete",
      %Schema{
        type: :string,
        title: "Status id",
        description: "Id of the deleted status",
        allOf: [FlakeID],
        example: "some-opaque-id"
      },
      payload_type: :string,
      has_stream: false
    )
  end

  defp pleroma_respond_event do
    server_sent_event_helper(
      "Server response",
      "A response to a client-sent event.",
      "pleroma:respond",
      %Schema{
        type: :object,
        title: "Results",
        required: [:result, :type],
        properties: %{
          result: %Schema{
            type: :string,
            title: "Result of the request",
            enum: ["success", "error", "ignored"]
          },
          error: %Schema{
            type: :string,
            title: "Error code",
            description: "An error identifier. Only appears if `result` is `error`."
          },
          type: %Schema{
            type: :string,
            description: "Type of the request."
          }
        },
        example: %{"result" => "success", "type" => "pleroma:authenticate"}
      },
      has_stream: false
    )
  end

  defp client_sent_events do
    %Schema{
      oneOf: [
        subscribe_event(),
        unsubscribe_event(),
        authenticate_event()
      ]
    }
  end

  defp request_body(description, schema, opts \\ []) do
    %OpenApiSpex.RequestBody{
      description: description,
      content: %{
        "application/json" => %OpenApiSpex.MediaType{
          schema: schema,
          example: opts[:example],
          examples: opts[:examples]
        }
      }
    }
  end

  defp client_sent_event_helper(name, description, type, properties, opts) do
    required = opts[:required] || []

    %Schema{
      type: :object,
      title: name,
      required: [:type] ++ required,
      description: description,
      properties:
        %{
          type: %Schema{type: :string, enum: [type], description: "Type of the event."}
        }
        |> Map.merge(properties),
      example: opts[:example]
    }
  end

  defp subscribe_event do
    client_sent_event_helper(
      "Subscribe",
      "Subscribe to a stream.",
      "subscribe",
      stream_specifier(),
      required: [:stream],
      example: %{"type" => "subscribe", "stream" => "list", "list" => "1"}
    )
  end

  defp unsubscribe_event do
    client_sent_event_helper(
      "Unsubscribe",
      "Unsubscribe from a stream.",
      "subscribe",
      stream_specifier(),
      required: [:stream],
      example: %{
        "type" => "unsubscribe",
        "stream" => "public:remote:media",
        "instance" => "example.org"
      }
    )
  end

  defp authenticate_event do
    client_sent_event_helper(
      "Authenticate",
      "Authenticate via an access token.",
      "pleroma:authenticate",
      %{
        token: %Schema{
          type: :string,
          description: "An OAuth access token with corresponding permissions.",
          example: "some token"
        }
      },
      required: [:token]
    )
  end

  defp stream_specifier do
    %{
      stream: %Schema{
        type: :string,
        description: "The name of the stream.",
        enum:
          Pleroma.Constants.public_streams() ++
            [
              "public:remote",
              "public:remote:media",
              "user",
              "user:pleroma_chat",
              "user:notification",
              "direct",
              "list",
              "hashtag"
            ]
      },
      list: %Schema{
        type: :string,
        title: "List id",
        description: "The id of the list. Required when `stream` is `list`.",
        example: "some-id"
      },
      tag: %Schema{
        type: :string,
        title: "Hashtag name",
        description: "The name of the hashtag. Required when `stream` is `hashtag`.",
        example: "mew"
      },
      instance: %Schema{
        type: :string,
        title: "Domain name",
        description:
          "Domain name of the instance. Required when `stream` is `public:remote` or `public:remote:media`.",
        example: "example.org"
      }
    }
  end
end