// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0

import 'dart:async';

import 'package:amplify_api_dart/src/graphql/helpers/send_graphql_request.dart';
import 'package:amplify_api_dart/src/graphql/providers/app_sync_api_key_auth_provider.dart';
import 'package:amplify_api_dart/src/graphql/providers/oidc_function_api_auth_provider.dart';
import 'package:amplify_api_dart/src/graphql/web_socket/blocs/web_socket_bloc.dart';
import 'package:amplify_api_dart/src/graphql/web_socket/services/web_socket_service.dart';
import 'package:amplify_api_dart/src/graphql/web_socket/state/web_socket_state.dart';
import 'package:amplify_api_dart/src/graphql/web_socket/types/connectivity_platform.dart';
import 'package:amplify_api_dart/src/util/amplify_api_config.dart';
import 'package:amplify_api_dart/src/util/amplify_authorization_rest_client.dart';
import 'package:amplify_core/amplify_core.dart';
import 'package:meta/meta.dart';

/// {@template amplify_api_dart.amplify_api_dart}
/// The AWS implementation of the Amplify API category in Dart.
/// {@endtemplate}
class AmplifyAPIDart extends APIPluginInterface with AWSDebuggable {
  /// {@macro amplify_api_dart.amplify_api_dart}
  AmplifyAPIDart({
    List<APIAuthProvider> authProviders = const [],
    ConnectivityPlatform connectivity = const ConnectivityPlatform(),
    AWSHttpClient? baseHttpClient,
    this.modelProvider,
    this.subscriptionOptions,
  })  : _baseHttpClient = baseHttpClient,
        _connectivity = connectivity {
    authProviders.forEach(registerAuthProvider);
  }

  late final AWSApiPluginConfig _apiConfig;
  final AWSHttpClient? _baseHttpClient;
  late final AmplifyAuthProviderRepository _authProviderRepo;

  /// Creates a stream representing network connectivity at the hardware level.
  final ConnectivityPlatform _connectivity;

  /// A map of the keys from the Amplify API config with auth modes to HTTP clients
  /// to use for requests to that endpoint/auth mode. e.g. { "myEndpoint.AWS_IAM": AWSHttpClient}
  final Map<String, AWSHttpClient> _clientPool = {};

  /// A map of the keys from the Amplify API config websocket connections to use
  /// for that endpoint.
  final Map<String, WebSocketBloc> _webSocketBlocPool = {};

  final StreamController<ApiHubEvent> _hubEventController =
      StreamController<ApiHubEvent>.broadcast();

  /// Subscription options
  final GraphQLSubscriptionOptions? subscriptionOptions;

  @override
  Future<void> reset() async {
    for (final bloc in _webSocketBlocPool.values) {
      bloc.add(const ShutdownEvent());
    }

    await Future.wait(
      _webSocketBlocPool.values.map((bloc) => bloc.done.future),
    );

    await _hubEventController.close();
    await super.reset();
  }

  @override
  Future<void> configure({
    AmplifyConfig? config,
    required AmplifyAuthProviderRepository authProviderRepo,
  }) async {
    final apiConfig = config?.api?.awsPlugin;
    if (apiConfig == null) {
      throw ConfigurationError(
        'No AWS API config found',
        recoverySuggestion: 'Add API from the Amplify CLI. See '
            'https://docs.amplify.aws/lib/graphqlapi/getting-started/q/platform/flutter/#configure-api',
      );
    }
    for (final entry in apiConfig.endpoints.entries) {
      if (!entry.value.endpoint.startsWith('https')) {
        throw ConfigurationError(
          'Non-HTTPS endpoint found for ${entry.key} which is not supported.',
          recoverySuggestion:
              'Ensure the configured endpoint for ${entry.key} utilizes https.',
        );
      }
    }
    _apiConfig = apiConfig;
    _authProviderRepo = authProviderRepo;
    _registerApiPluginAuthProviders();
    Amplify.Hub.addChannel(HubChannel.Api, _hubEventController.stream);
  }

  /// Register AmplifyAuthProviders that are specific to API category (API key,
  /// OIDC or Lambda).
  ///
  /// If an endpoint has an API key, ensure valid auth provider registered.
  ///
  /// Register OIDC/Lambda set to _authProviders in constructor.
  void _registerApiPluginAuthProviders() {
    _apiConfig.endpoints.forEach((key, value) {
      // Check the presence of apiKey (not auth type) because other modes might
      // have a key if not the primary auth mode.
      if (value.apiKey != null) {
        _authProviderRepo.registerAuthProvider(
          APIAuthorizationType.apiKey.authProviderToken,
          AppSyncApiKeyAuthProvider(),
        );
      }
    });

    // Register OIDC/Lambda auth providers.
    for (final authProvider in authProviders.values) {
      _authProviderRepo.registerAuthProvider(
        authProvider.type.authProviderToken,
        OidcFunctionAuthProvider(authProvider),
      );
    }
  }

  // TODO(equartey): add [apiName] to event to distinguished when multiple blocs are running.
  void _emitHubEvent(WebSocketState state) {
    if (state is ConnectingState || state is ReconnectingState) {
      _hubEventController.add(SubscriptionHubEvent.connecting());
    } else if (state is ConnectedState) {
      _hubEventController.add(SubscriptionHubEvent.connected());
    } else if (state is PendingDisconnect) {
      _hubEventController.add(SubscriptionHubEvent.pendingDisconnect());
    } else if (state is DisconnectedState) {
      _hubEventController.add(SubscriptionHubEvent.disconnected());
    } else if (state is FailureState) {
      _hubEventController.add(SubscriptionHubEvent.failed());
    }
  }

  /// Returns the HTTP client to be used for REST/GraphQL operations.
  ///
  /// Use [apiName] if there are multiple endpoints of the same type.
  @visibleForTesting
  AWSHttpClient getHttpClient(
    EndpointType type, {
    String? apiName,
    APIAuthorizationType? authorizationMode,
  }) {
    final endpoint = _apiConfig.getEndpoint(
      type: type,
      apiName: apiName,
    );
    final authModeForClientKey =
        authorizationMode ?? endpoint.config.authorizationType;
    final clientPoolKey = '${endpoint.name}.${authModeForClientKey.name}';
    return _clientPool[clientPoolKey] ??= AmplifyHttpClient(
      dependencies,
      baseClient: AmplifyAuthorizationRestClient(
        endpointConfig: endpoint.config,
        baseClient: _baseHttpClient ?? dependencies.getOrCreate(),
        authorizationMode: authorizationMode,
        authProviderRepo: _authProviderRepo,
      ),
    );
  }

  WebSocketBloc _webSocketBloc({String? apiName}) {
    final endpoint = _apiConfig.getEndpoint(
      type: EndpointType.graphQL,
      apiName: apiName,
    );

    return _webSocketBlocPool[endpoint.name] ??= createWebSocketBloc(endpoint)
      ..stream.listen((event) {
        _emitHubEvent(event);
        if (event is PendingDisconnect) {
          _webSocketBlocPool.remove(endpoint.name);
        }
      });
  }

  /// Returns the websocket bloc to use for a given endpoint.
  ///
  /// Use [endpoint] if there are multiple endpoints.
  @visibleForTesting
  WebSocketBloc createWebSocketBloc(EndpointConfig endpoint) {
    return WebSocketBloc(
      config: endpoint.config,
      authProviderRepo: _authProviderRepo,
      wsService: AmplifyWebSocketService(),
      subscriptionOptions:
          subscriptionOptions ?? const GraphQLSubscriptionOptions(),
      connectivity: _connectivity,
    );
  }

  Uri _getGraphQLUri(String? apiName) {
    final endpoint = _apiConfig.getEndpoint(
      type: EndpointType.graphQL,
      apiName: apiName,
    );
    return endpoint.getUri();
  }

  Uri _getRestUri(
    String path,
    String? apiName,
    Map<String, dynamic>? queryParameters,
  ) {
    final endpoint = _apiConfig.getEndpoint(
      type: EndpointType.rest,
      apiName: apiName,
    );
    return endpoint.getUri(path: path, queryParameters: queryParameters);
  }

  @override
  final ModelProviderInterface? modelProvider;

  // ====== GraphQL ======

  @override
  GraphQLOperation<T> query<T>({required GraphQLRequest<T> request}) {
    final graphQLClient = getHttpClient(
      EndpointType.graphQL,
      apiName: request.apiName,
      authorizationMode: request.authorizationMode,
    );
    final uri = _getGraphQLUri(request.apiName);

    return sendGraphQLRequest<T>(
      request: request,
      client: graphQLClient,
      uri: uri,
    );
  }

  @override
  GraphQLOperation<T> mutate<T>({required GraphQLRequest<T> request}) {
    final graphQLClient = getHttpClient(
      EndpointType.graphQL,
      apiName: request.apiName,
      authorizationMode: request.authorizationMode,
    );
    final uri = _getGraphQLUri(request.apiName);

    return sendGraphQLRequest<T>(
      request: request,
      client: graphQLClient,
      uri: uri,
    );
  }

  @override
  Stream<GraphQLResponse<T>> subscribe<T>(
    GraphQLRequest<T> request, {
    void Function()? onEstablished,
  }) {
    final event = SubscribeEvent(request, onEstablished);
    return _webSocketBloc(apiName: request.apiName).subscribe(event);
  }

  // ====== REST =======

  @override
  RestOperation delete(
    String path, {
    HttpPayload? body,
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSStreamedHttpRequest.delete(
        uri,
        body: body,
        headers: headers,
      ).send(client: client),
    );
  }

  @override
  RestOperation get(
    String path, {
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSHttpRequest.get(
        uri,
        headers: headers,
      ).send(client: client),
    );
  }

  @override
  RestOperation head(
    String path, {
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSHttpRequest.head(
        uri,
        headers: headers,
      ).send(client: client),
    );
  }

  @override
  RestOperation patch(
    String path, {
    HttpPayload? body,
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSStreamedHttpRequest.patch(
        uri,
        headers: headers,
        body: body ?? const HttpPayload.empty(),
      ).send(client: client),
    );
  }

  @override
  RestOperation post(
    String path, {
    HttpPayload? body,
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSStreamedHttpRequest.post(
        uri,
        headers: headers,
        body: body ?? const HttpPayload.empty(),
      ).send(client: client),
    );
  }

  @override
  RestOperation put(
    String path, {
    HttpPayload? body,
    Map<String, String>? headers,
    Map<String, String>? queryParameters,
    String? apiName,
  }) {
    final uri = _getRestUri(path, apiName, queryParameters);
    final client = getHttpClient(EndpointType.rest, apiName: apiName);
    return RestOperation.fromHttpOperation(
      AWSStreamedHttpRequest.put(
        uri,
        headers: headers,
        body: body ?? const HttpPayload.empty(),
      ).send(client: client),
    );
  }

  @override
  String get runtimeTypeName => 'AmplifyAPIDart';
}