// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package admin

import (
	"context"

	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
)

// Clusters is admin interface for clusters management
type Clusters interface {
	// List returns the list of clusters
	List() ([]string, error)

	// ListWithContext returns the list of clusters
	ListWithContext(context.Context) ([]string, error)

	// Get the configuration data for the specified cluster
	Get(string) (utils.ClusterData, error)

	// GetWithContext returns the configuration data for the specified cluster
	GetWithContext(context.Context, string) (utils.ClusterData, error)

	// Create a new cluster
	Create(utils.ClusterData) error

	// CreateWithContext creates a new cluster
	CreateWithContext(context.Context, utils.ClusterData) error

	// Delete an existing cluster
	Delete(string) error

	// DeleteWithContext deletes an existing cluster
	DeleteWithContext(context.Context, string) error

	// Update the configuration for a cluster
	Update(utils.ClusterData) error

	// UpdateWithContext updates the configuration for a cluster
	UpdateWithContext(context.Context, utils.ClusterData) error

	// UpdatePeerClusters updates peer cluster names.
	UpdatePeerClusters(string, []string) error

	// UpdatePeerClustersWithContext updates peer cluster names.
	UpdatePeerClustersWithContext(context.Context, string, []string) error

	// GetPeerClusters returns peer-cluster names
	GetPeerClusters(string) ([]string, error)

	// GetPeerClusters returns peer-cluster names
	GetPeerClustersWithContext(context.Context, string) ([]string, error)

	// CreateFailureDomain creates a domain into cluster
	CreateFailureDomain(utils.FailureDomainData) error

	// CreateFailureDomain creates a domain into cluster
	CreateFailureDomainWithContext(context.Context, utils.FailureDomainData) error

	// GetFailureDomain returns the domain registered into a cluster
	GetFailureDomain(clusterName, domainName string) (utils.FailureDomainData, error)

	// GetFailureDomain returns the domain registered into a cluster
	GetFailureDomainWithContext(ctx context.Context, clusterName, domainName string) (utils.FailureDomainData, error)

	// ListFailureDomains returns all registered domains in cluster
	ListFailureDomains(string) (utils.FailureDomainMap, error)

	// ListFailureDomains returns all registered domains in cluster
	ListFailureDomainsWithContext(context.Context, string) (utils.FailureDomainMap, error)

	// DeleteFailureDomain deletes a domain in cluster
	DeleteFailureDomain(utils.FailureDomainData) error

	// DeleteFailureDomain deletes a domain in cluster
	DeleteFailureDomainWithContext(context.Context, utils.FailureDomainData) error

	// UpdateFailureDomain updates a domain into cluster
	UpdateFailureDomain(utils.FailureDomainData) error

	// UpdateFailureDomain updates a domain into cluster
	UpdateFailureDomainWithContext(context.Context, utils.FailureDomainData) error
}

type clusters struct {
	pulsar   *pulsarClient
	basePath string
}

// Clusters is used to access the cluster endpoints.
func (c *pulsarClient) Clusters() Clusters {
	return &clusters{
		pulsar:   c,
		basePath: "/clusters",
	}
}

func (c *clusters) List() ([]string, error) {
	return c.ListWithContext(context.Background())
}

func (c *clusters) ListWithContext(ctx context.Context) ([]string, error) {
	var clusters []string
	err := c.pulsar.Client.GetWithContext(ctx, c.pulsar.endpoint(c.basePath), &clusters)
	return clusters, err
}

func (c *clusters) Get(name string) (utils.ClusterData, error) {
	return c.GetWithContext(context.Background(), name)
}

func (c *clusters) GetWithContext(ctx context.Context, name string) (utils.ClusterData, error) {
	cdata := utils.ClusterData{}
	endpoint := c.pulsar.endpoint(c.basePath, name)
	err := c.pulsar.Client.GetWithContext(ctx, endpoint, &cdata)
	return cdata, err
}

func (c *clusters) Create(cdata utils.ClusterData) error {
	return c.CreateWithContext(context.Background(), cdata)
}

func (c *clusters) CreateWithContext(ctx context.Context, cdata utils.ClusterData) error {
	endpoint := c.pulsar.endpoint(c.basePath, cdata.Name)
	return c.pulsar.Client.PutWithContext(ctx, endpoint, &cdata)
}

func (c *clusters) Delete(name string) error {
	return c.DeleteWithContext(context.Background(), name)
}

func (c *clusters) DeleteWithContext(ctx context.Context, name string) error {
	endpoint := c.pulsar.endpoint(c.basePath, name)
	return c.pulsar.Client.DeleteWithContext(ctx, endpoint)
}

func (c *clusters) Update(cdata utils.ClusterData) error {
	return c.UpdateWithContext(context.Background(), cdata)
}

func (c *clusters) UpdateWithContext(ctx context.Context, cdata utils.ClusterData) error {
	endpoint := c.pulsar.endpoint(c.basePath, cdata.Name)
	return c.pulsar.Client.PostWithContext(ctx, endpoint, &cdata)
}

func (c *clusters) GetPeerClusters(name string) ([]string, error) {
	return c.GetPeerClustersWithContext(context.Background(), name)
}

func (c *clusters) GetPeerClustersWithContext(ctx context.Context, name string) ([]string, error) {
	var peerClusters []string
	endpoint := c.pulsar.endpoint(c.basePath, name, "peers")
	err := c.pulsar.Client.GetWithContext(ctx, endpoint, &peerClusters)
	return peerClusters, err
}

func (c *clusters) UpdatePeerClusters(cluster string, peerClusters []string) error {
	return c.UpdatePeerClustersWithContext(context.Background(), cluster, peerClusters)
}

func (c *clusters) UpdatePeerClustersWithContext(ctx context.Context, cluster string, peerClusters []string) error {
	endpoint := c.pulsar.endpoint(c.basePath, cluster, "peers")
	return c.pulsar.Client.PostWithContext(ctx, endpoint, peerClusters)
}

func (c *clusters) CreateFailureDomain(data utils.FailureDomainData) error {
	return c.CreateFailureDomainWithContext(context.Background(), data)
}

func (c *clusters) CreateFailureDomainWithContext(ctx context.Context, data utils.FailureDomainData) error {
	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
	return c.pulsar.Client.PostWithContext(ctx, endpoint, &data)
}

func (c *clusters) GetFailureDomain(clusterName string, domainName string) (utils.FailureDomainData, error) {
	return c.GetFailureDomainWithContext(context.Background(), clusterName, domainName)
}

func (c *clusters) GetFailureDomainWithContext(
	ctx context.Context,
	clusterName string,
	domainName string,
) (utils.FailureDomainData, error) {
	var res utils.FailureDomainData
	endpoint := c.pulsar.endpoint(c.basePath, clusterName, "failureDomains", domainName)
	err := c.pulsar.Client.GetWithContext(ctx, endpoint, &res)
	return res, err
}

func (c *clusters) ListFailureDomains(
	clusterName string,
) (utils.FailureDomainMap, error) {
	return c.ListFailureDomainsWithContext(context.Background(), clusterName)
}

func (c *clusters) ListFailureDomainsWithContext(
	ctx context.Context,
	clusterName string,
) (utils.FailureDomainMap, error) {
	var domainData utils.FailureDomainMap
	endpoint := c.pulsar.endpoint(c.basePath, clusterName, "failureDomains")
	err := c.pulsar.Client.GetWithContext(ctx, endpoint, &domainData)
	return domainData, err
}

func (c *clusters) DeleteFailureDomain(data utils.FailureDomainData) error {
	return c.DeleteFailureDomainWithContext(context.Background(), data)
}

func (c *clusters) DeleteFailureDomainWithContext(ctx context.Context, data utils.FailureDomainData) error {
	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
	return c.pulsar.Client.DeleteWithContext(ctx, endpoint)
}

func (c *clusters) UpdateFailureDomain(data utils.FailureDomainData) error {
	return c.UpdateFailureDomainWithContext(context.Background(), data)
}

func (c *clusters) UpdateFailureDomainWithContext(ctx context.Context, data utils.FailureDomainData) error {
	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
	return c.pulsar.Client.PostWithContext(ctx, endpoint, &data)
}
