/*
 * 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/clause"

	"gorm.io/gorm"
)

const (
	IngressTableName = "ingress"
)

type Ingress 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:ingress_type,priority:3;NOT NULL;DEFAULT:''"`
	Name        string `gorm:"column:name;type:varchar(64);uniqueIndex:ingress_type,priority:2;NOT NULL;DEFAULT:''"`
	Endpoint    string `gorm:"column:endpoint;type:varchar(512);NOT NULL;DEFAULT:''"`
	IngressType string `gorm:"column:ingress_type;type:enum('kafka','sqs');uniqueIndex:ingress_type,priority:1;NOT NULL;DEFAULT:'sqs'"`

	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 (in *Ingress) TableName() string {
	return "ingress"
}

type IngressDao interface {
	Insert(ctx context.Context, ingress *Ingress) error
	Find(ctx context.Context, userID string, ingress_name string) (*Ingress, bool, error)
	Delete(ctx context.Context, userId string, ingressName string, ingress_type string) error
	List(ctx context.Context, cond *ListIngressCond) ([]Ingress, error)
}

type ingressDao struct {
	db *gorm.DB
}

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

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

func (dao *ingressDao) Delete(ctx context.Context, userId string, ingressName string, ingress_type string) error {
	ingress := new(Ingress)
	return dao.db.WithContext(ctx).
		Table(ingress.TableName()).Where(
		"user_id = ? && name = ? && ingress_type = ?", userId, ingressName, ingress_type).Delete(Ingress{}).Error
}

func (dao *ingressDao) Insert(ctx context.Context, ingress *Ingress) error {
	return dao.db.WithContext(ctx).
		Table(ingress.TableName()).
		Clauses(clause.OnConflict{
			Columns: []clause.Column{
				{Name: "user_id"},
				{Name: "name"},
				{Name: "ingress_type"},
			},
			UpdateAll: true,
		}).Create(ingress).Error
}

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

type ListIngressCond struct {
	UserID       string
	ReverseOrder bool
	Limit        int
}

func (dao *ingressDao) List(ctx context.Context, cond *ListIngressCond) ([]Ingress, error) {
	q := dao.db.WithContext(ctx).Table(IngressTableName)
	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([]Ingress, 0)
	for rows.Next() {
		ingress := Ingress{}
		if err := dao.db.ScanRows(rows, &ingress); err != nil {
			return nil, err
		}
		data = append(data, ingress)
	}
	return data, nil
}