/*
 * 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 service

import (
	"context"
	"fmt"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/mimir/errors"
	"github.com/pingcap/mimir/model"
	"github.com/pingcap/mimir/pkg/event"
	"github.com/pingcap/mimir/pkg/event/event_bus"
	"github.com/pingcap/mimir/pkg/protos"
)

func SinkDaoOpt(dao model.SinkDao) Option {
	return func(service *apiService) error {
		service.sinkDao = dao
		return nil
	}
}

// CreateSink use the param to create a sink record in database
func (svc *apiService) CreateSink(ctx context.Context, param *protos.CreateSinkParam) (*protos.CreateSinkData, error) {
	// Post event to event bus first
	if err := event_bus.EventBus.PostSync(event.AddSinkEvent{
		Database:  param.Database,
		TableName: param.Table,
		Schedule:  param.Schedule,
	}); err != nil {
		log.Error("prepare sink error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("prepare sink %s failed", param.SinkID))
	}
	sink := &model.Sink{
		UserID:   protos.DefaultUserID,
		SinkID:   param.SinkID,
		SinkType: model.SinkTypeTiDB, // at this time, only tidb is supprted
		Table:    param.Table,
		Database: param.Database,
		Schedule: param.Schedule,
	}
	if err := svc.sinkDao.Insert(ctx, sink); err != nil {
		log.Error("insert sink error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("insert sink %s failed", param.SinkID))
	}
	return &protos.CreateSinkData{
		SinkID: param.SinkID,
	}, nil
}

func (svc *apiService) ListSink(ctx context.Context) (*protos.ListSinkData, error) {
	// TODO: Refactor the hardcode userID.
	userID := protos.DefaultUserID
	// TODO: Refactor the hardcode Limit.
	cond := model.ListSinkCond{
		UserID: userID,
		Limit:  100,
	}
	sinks, err := svc.sinkDao.List(ctx, &cond)
	if err != nil {
		log.Error("list sink error", zap.Error(err))
		return nil, errors.ErrInternal("list sink failed, caused by dao connection")
	}
	data := &protos.ListSinkData{Items: []protos.ListSinkItem{}}
	for _, sink := range sinks {
		data.Items = append(data.Items, protos.ListSinkItem{
			UserID:   sink.UserID,
			SinkID:   sink.SinkID,
			SinkType: "tidb", // only tidb is supported now
			Table:    sink.Table,
		})
	}
	return data, nil
}

func (svc *apiService) RmSink(ctx context.Context, param *protos.RmSinkParam) (*protos.RmSinkData, error) {
	isExist, err := svc.CheckExist(ctx, "sink_id", param.SinkID)
	if err != nil {
		log.Error(fmt.Sprintf("check exist failed, %s", err))
		return nil, err
	}
	if isExist {
		return nil, fmt.Errorf("please rm related job first")
	}
	sink, exist, err := svc.sinkDao.Find(ctx, protos.DefaultUserID, param.SinkID)
	if err != nil {
		log.Error(err.Error())
		return nil, errors.ErrInternal(err.Error())
	}
	if !exist {
		return nil, errors.ErrNotFound("sink not found")
	}

	// Post event to event bus first
	if err := event_bus.EventBus.PostSync(event.DeleteSinkEvent{
		Database:  sink.Database,
		TableName: sink.Table,
	}); err != nil {
		log.Error("prepare remove sink error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("prepare sink %s failed: %s", param.SinkID, err))
	}

	err = svc.sinkDao.Delete(ctx, protos.DefaultUserID, param.SinkID)
	if err != nil {
		log.Error("delete sink", zap.Error(err))
		return nil, errors.ErrInternal(err.Error())
	}
	return &protos.RmSinkData{SinkID: sink.SinkID}, nil
}