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

import (
	"context"
	"errors"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type SinkType int

const (
	SinkTableName = "sink"

	SinkTypeTiDB SinkType = 1
)

type Sink struct {
	ID       uint     `gorm:"column:id;type:int(11) unsigned NOT NULL AUTO_INCREMENT;PRIMARY_KEY"`
	UserID   string   `gorm:"column:user_id;type:varchar(64);uniqueIndex:sink_user_id_sink_id_index,priority:2;NOT NULL;DEFAULT:''"`
	SinkID   string   `gorm:"column:sink_id;type:varchar(64);uniqueIndex:sink_user_id_sink_id_index,priority:1;NOT NULL;DEFAULT:''"`
	SinkType SinkType `gorm:"column:sink_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
	Table    string   `gorm:"column:table;type:varchar(128);NOT NULL;DEFAULT:''"`
	Database string   `gorm:"column:database;type:varchar(128);NOT NULL;DEFAULT:''"`
	Schedule string   `gorm:"column:schedule;type:varchar(128);NOT NULL;DEFAULT:''"`

	// Disallow gorm to set the `created_at`
	CreatedAt time.Time `gorm:"column:created_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP"`
	UpdatedAt time.Time `gorm:"column:updated_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"`
}

func (s *Sink) TableName() string {
	return SinkTableName
}

type SinkDao interface {
	Insert(ctx context.Context, sink *Sink) error
	List(ctx context.Context, cond *ListSinkCond) ([]Sink, error)
	Update(ctx context.Context, sink *Sink) error
	Find(ctx context.Context, userID string, sinkID string) (*Sink, bool, error)
	Delete(ctx context.Context, userID string, sinkID string) error
}

type sinkDao struct {
	db *gorm.DB
}

func NewSinkDao(options ...DaoOption) (SinkDao, error) {
	dao := new(sinkDao)
	for _, option := range options {
		if err := option(dao); err != nil {
			return nil, err
		}
	}
	return dao, nil
}

func (dao *sinkDao) SetDB(db *gorm.DB) {
	dao.db = db
}

func (dao *sinkDao) Insert(ctx context.Context, sink *Sink) error {
	return dao.db.WithContext(ctx).
		Table(sink.TableName()).
		Clauses(clause.OnConflict{
			Columns: []clause.Column{
				{Name: "user_id"},
				{Name: "sink_id"},
			},
			UpdateAll: true,
		}).Create(sink).Error
}

type ListSinkCond struct {
	UserID       string
	ReverseOrder bool
	Limit        int
}

func (dao *sinkDao) List(ctx context.Context, cond *ListSinkCond) ([]Sink, error) {
	q := dao.db.WithContext(ctx).Table(SinkTableName)
	if len(cond.UserID) != 0 {
		q = q.Where("user_id = ?", cond.UserID)
	}
	if cond.ReverseOrder {
		q = q.Order("created_at DESC")
	}
	if cond.Limit != 0 {
		q.Limit(cond.Limit)
	}
	rows, err := q.Rows()
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	data := make([]Sink, 0)
	for rows.Next() {
		sink := Sink{}
		if err := dao.db.ScanRows(rows, &sink); err != nil {
			return nil, err
		}
		data = append(data, sink)
	}
	return data, nil
}

func (dao *sinkDao) Find(ctx context.Context, userID string, sinkID string) (*Sink, bool, error) {
	sink := new(Sink)
	if err := dao.db.WithContext(ctx).Table(sink.TableName()).Where("user_id = ? && sink_id = ?", userID, sinkID).First(sink).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, false, nil
		}
		return nil, false, err
	}
	return sink, true, nil
}

func (dao *sinkDao) Update(ctx context.Context, sink *Sink) error {
	return dao.db.WithContext(ctx).Save(sink).Error
}

func (dao *sinkDao) Delete(ctx context.Context, userID string, sinkID string) error {
	if err := dao.db.WithContext(ctx).Table(SinkTableName).Where(
		"user_id = ? && sink_id = ?", userID, sinkID).Delete(&Sink{}).Error; err != nil {
		return err
	}
	return nil
}