/*
 * Copyright 2021 PingCAP, Inc.
 *
 * Licensed 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,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package client

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"

	"github.com/pingcap/mimir/pkg/protos"
)

func (c *Client) CreateSink(ctx context.Context, param *protos.CreateSinkParam) (*protos.CreateSinkData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v2/sink/create", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.CreateSinkData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) ListSink(ctx context.Context) (*protos.ListSinkData, error) {
	req, err := http.NewRequest("GET", c.endpoint+"/api/v2/sink/list", nil)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.ListSinkData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) RmSink(ctx context.Context, param *protos.RmSinkParam) (*protos.RmSinkData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v2/sink/rm", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.RmSinkData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}