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

import (
	"context"
	"crypto/tls"
	"flag"
	"log"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/scram"

	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sqs"

	"golang.org/x/sync/errgroup"
)

func main() {

	var dsn string
	var sourceType string
	var tableName string

	// sqs related value
	var sqsURL string

	// kafka related value
	var topic string
	var broker string
	var groupID string

	// optional
	var skipErr bool
	var username string
	var password string

	flag.StringVar(&dsn, "dsn", "root:@(127.0.0.1:4000)/mimir?charset=utf8mb4&parseTime=True&loc=UTC&timeout=30s&readTimeout=30s&writeTimeout=30s", "mysql dsn configuration")
	flag.StringVar(&tableName, "table", "", "table name")
	flag.StringVar(&sourceType, "source-type", "", "source type, available value is sqs|kafka")
	flag.StringVar(&sqsURL, "sqs-url", "", "sqs url")

	flag.StringVar(&broker, "broker", "", "kafka broker list, seperated by ,")
	flag.StringVar(&topic, "topic", "", "kafka topic")
	flag.StringVar(&groupID, "group-id", "", "kafka consumer group id")
	flag.BoolVar(&skipErr, "skip-error", false, "set consumer to skip err, default is false")
	flag.StringVar(&username, "username", "", "username used to connect kafka")
	flag.StringVar(&password, "password", "", "password used to connect kafka")
	flag.Parse()

	// mapper
	db, err := Dial(&MysqlConfig{DSN: dsn})
	if err != nil {
		log.Fatalf("dial database with dsn %+v error %+v", dsn, err)
	}
	mapper, err := NewDefaultMapper(db, tableName)
	if err != nil {
		log.Fatalf("new default mapper err: %+v", err)
	}
	var consumer IQueueConsumer
	switch sourceType {
	case "sqs":
		if len(sqsURL) == 0 {
			log.Fatalf("sqs url is empty")
		}
		sess := session.Must(session.NewSessionWithOptions(session.Options{
			SharedConfigState: session.SharedConfigEnable,
		}))

		c, err := NewSQSConsumer(sqs.New(sess), sqsURL,
			WithVisibilityTimeout(20),
		)
		if err != nil {
			log.Fatalf("new sqs consumer error %+v", err)
		}
		c.SetMapper(mapper)
		c.SkipError(skipErr)
		consumer = c
	case "kafka":
		if len(broker) == 0 {
			log.Fatal("kafka broker is empty")
		}
		if len(topic) == 0 {
			log.Fatal("kafka topic is empty")
		}
		if len(groupID) == 0 {
			log.Fatal("kafka group id is empty")
		}
		dialer := &kafka.Dialer{
			Timeout:   3 * time.Second,
			DualStack: true,
			TLS:       &tls.Config{InsecureSkipVerify: true},
		}
		if len(username) > 0 && len(password) > 0 {
			mechanism, err := scram.Mechanism(scram.SHA512, username, password)
			if err != nil {
				log.Fatalf("new mechanism error: %+v", err)
			}
			dialer.SASLMechanism = mechanism
		}
		c, err := NewKafkaConsumer(dialer, strings.Split(broker, ","), topic, groupID)
		if err != nil {
			log.Fatalf("new kafka consumer error %+v", err)
		}
		c.SetMapper(mapper)
		c.SkipError(skipErr)
		consumer = c
	default:
		log.Fatalf("source type %+v is not supported\n", sourceType)
	}
	sigCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	errG, ctx := errgroup.WithContext(sigCtx)
	errG.Go(func() error {
		return consumer.Start(ctx)
	})
	if err := errG.Wait(); err != nil {
		stop()
		log.Fatalf("error group err: %+v", err)
	}
	stop()
}