package main

import (
	"compress/gzip"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	geoip2 "github.com/oschwald/geoip2-golang"
	"github.com/panjf2000/ants"
	"github.com/qiniu/go-sdk/v7/auth"
	"github.com/qiniu/go-sdk/v7/cdn"
	"github.com/robfig/cron/v3"
	"github.com/spf13/cast"
	"github.com/spf13/viper"
	"gorm.io/gorm"

	"github.com/pingcap/lark2es/pkg"
	"github.com/pingcap/lark2es/pkg/config"
	"github.com/pingcap/lark2es/pkg/db"
	"github.com/pingcap/lark2es/pkg/job"
	"github.com/pingcap/lark2es/pkg/util"
)

type CdnDownloadLogV1 struct {
	Date         string
	Component    string
	Source       string
	IP           string
	Country      string
	City         string
	RawData      string
	CDN          string
	ResponseTime int
	Key          string
	LogTime      string
	Version      string
	State        string
	Size         int
	Code         int
	FromGithub   bool
	PullDate string
}

type ISource interface {
	dealer(content string, ops ...func(log *CdnDownloadLogV1)) []CdnDownloadLogV1
	fetcher(source string) (string, error)
	cron.Job
}

func queryLocationByIp(ip string) (countryName string, cityName string) {
	city, err := geoIPReader.City(net.ParseIP(ip))
	if err != nil {
		return
	}

	cityName = city.City.Names["en"]
	countryName = city.Country.Names["en"]
	return
}

func Save(logs []CdnDownloadLogV1, gdb *gorm.DB) {
	step := 1000
	idx := 0
	for idx = 0; idx+step < len(logs); idx = idx + step {
		err := gdb.Save(logs[idx : idx+step]).Error
		if err != nil {
			log.Println("[error] DB Save", err)
		}
	}
	if idx < len(logs) {
		err := gdb.Save(logs[idx:]).Error
		if err != nil {
			log.Println("[error] DB Save", err)
		}
	}
}

var (
	geoIPReader    *geoip2.Reader
	workPool       *ants.Pool
	versionRegex   *regexp.Regexp
	githubCidrConf GithubCIDR
	awsAfter       string
	checkCIRDs     []*net.IPNet
)

type GithubCIDR struct {
	VerifiablePasswordAuthentication bool `json:"verifiable_password_authentication"`
	SSHKeyFingerprints struct {
		SHA256RSA string `json:"SHA256_RSA"`
		SHA256ECDSA string `json:"SHA256_ECDSA"`
		SHA256ED25519 string `json:"SHA256_ED25519"`
	} `json:"ssh_key_fingerprints"`
	SSHKeys []string `json:"ssh_keys"`
	Hooks []string `json:"hooks"`
	Web []string `json:"web"`
	API []string `json:"api"`
	Git []string `json:"git"`
	Packages []string `json:"packages"`
	Pages []string `json:"pages"`
	Importer []string `json:"importer"`
	Actions []string `json:"actions"`
	Dependabot []string `json:"dependabot"`
}

func init() {
	var err error
	geoIPReader, err = geoip2.Open("./GeoIP2-City.mmdb")
	if err != nil {
		log.Fatal(err)
	}
	workPool, _ = ants.NewPool(100)
	versionRegex = regexp.MustCompile(`.*-(?P<Version>v\d+\.\d+\.\d+).*`)

	req, err := pkg.GetRequest("https://api.github.com/meta", nil, nil, nil)
	pkg.Do(req, &githubCidrConf)
	checkCIRDs = make([]*net.IPNet, 0, len(githubCidrConf.Actions))
	for _, cird := range githubCidrConf.Actions {
		_, ipNet, _ := net.ParseCIDR(cird)
		checkCIRDs = append(checkCIRDs, ipNet)
	}
	log.Println("cird size ", len(checkCIRDs))
}

const internalip = `"45.251.23.186","180.167.192.138","117.50.84.111","106.75.116.22","106.75.96.68","128.1.49.135","106.75.52.157","118.194.230.143","106.75.64.135","203.90.236.205","117.50.66.239","40.74.68.232","106.75.54.70","118.113.15.98"`

func checkIsFromGithubAction(ip string) bool {
	contain := false
	for _, cird := range checkCIRDs {
		if cird.Contains(net.ParseIP(ip)) {
			contain = true
			break
		}
	}

	return contain
}

func isInternalIp(ip string) bool {
	return strings.Index(internalip, ip) != -1
}

func ipResolve(log *CdnDownloadLogV1) {
	log.Country, log.City = queryLocationByIp(log.IP)
}

func ipFromGithub(log *CdnDownloadLogV1) {
	log.FromGithub = checkIsFromGithubAction(log.IP)
}

func version(log *CdnDownloadLogV1) {
	matches := versionRegex.FindStringSubmatch(log.Component)
	if len(matches) == 2 {
		log.Version = matches[1]
	}
}

type QiniuSource struct {
	ak, sk string
	day    string
	cdn    string
	urls   []string
	db     *gorm.DB
}

func NewQiniuSource(ak, sk, cdn string, urls []string, db *gorm.DB) *QiniuSource {
	return &QiniuSource{ak: ak, sk: sk, urls: urls, db: db, cdn: cdn}
}

func (s *QiniuSource) SetDay(day string) *QiniuSource {
	s.day = day
	return s
}

func (s *QiniuSource) Run() {
	credential := auth.New(s.ak, s.sk)
	cdnManager := cdn.NewCdnManager(credential)
	if s.day == "" {
		s.day = time.Now().AddDate(0, 0, -1).Format("2006-01-02")
	}
	defer func() {
		s.day = ""
	}()
	retryCt := 0
retry:
	resp, err := cdnManager.GetCdnLogList(s.day, s.urls)
	if err != nil && err.Error() != "get log list error, 200 成功" {
		if resp.Code == 400032 {
			time.Sleep(time.Second)
			retryCt++
			if retryCt > 8 {
				// todo cjl notify lark
				log.Println("[error] GetCdnLogList ", err.Error(), resp)
				return
			}
			goto retry
		}

		log.Println("[error] GetCdnLogList ", err.Error(), resp)
		return
	}

	for i := range s.urls {
		url := s.urls[i]
		if cdnlogs, ok := resp.Data[url]; ok {
			for _, cdnlog := range cdnlogs {
				content, err := s.fetcher(cdnlog.URL)
				if err != nil {
					log.Println("[error] fetcher", err, cdnlog.URL)
					continue
				}

				logs := s.dealer(content, func(log *CdnDownloadLogV1) {
					log.Source = url
					log.Date = s.day
				}, ipResolve, version)
				Save(logs, s.db)
			}
		}
	}
}

func (s *QiniuSource) dealer(content string, ops ...func(log *CdnDownloadLogV1)) []CdnDownloadLogV1 {
	lines := strings.Split(content, "\n")
	logs := make([]CdnDownloadLogV1, 0, len(lines))
	nowStr := time.Now().Format("2006-01-02 15:04:05")
	for _, line := range lines {
		if strings.Index(line, ".tar.gz") != -1 {
			lineSplits := strings.Split(line, " ")
			ip := lineSplits[0]
			filename := filepath.Base(lineSplits[6])
			if strings.Index(filename, ".tar.gz") != -1 && !isInternalIp(ip) {
				log := CdnDownloadLogV1{Component: filename, IP: ip,
					RawData: line, ResponseTime: cast.ToInt(lineSplits[2]),
					LogTime: nowStr, State: lineSplits[1], CDN: s.cdn,
					Size: cast.ToInt(lineSplits[9]), Code: cast.ToInt(lineSplits[8]),
				}
				for _, op := range ops {
					op(&log)
				}
				logs = append(logs, log)
			}
		}
	}

	return logs
}

func (s *QiniuSource) fetcher(source string) (string, error) {
	resp, err := http.Get(source)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	gzReader, err := gzip.NewReader(resp.Body)
	if err != nil {
		return "", err
	}

	data, err := ioutil.ReadAll(gzReader)
	if err != nil {
		return "", err
	}

	return util.ByteSliceToString(data), nil
}

type AWSS3Source struct {
	init bool
	after  string
	bucket string
	region string
	cdn    string
	client *s3.S3
	db     *gorm.DB
}

func newAWSS3Source(bucket, cdn, region string, db *gorm.DB) *AWSS3Source {
	s := &AWSS3Source{bucket: bucket, cdn: cdn, db: db, region: region}
	return s
}

func (s *AWSS3Source) fetcher(source string) (string, error) {
	objInput := s3.GetObjectInput{}
	objInput.SetBucket(s.bucket)
	objInput.SetKey(source)
	retryct := 0

s3retry:
	obj, err := s.client.GetObject(&objInput)
	if err != nil {
		if retryct <= 3 {
			retryct++
			goto s3retry
		}
		log.Println("[error] GetObject", err, s.bucket, source)
		return "", err
	}
	defer obj.Body.Close()

	gzipBodyReader, err := gzip.NewReader(obj.Body)
	if err != nil {
		log.Println("[error] gzip.NewReader", err, s.bucket, source)
		return "", err
	}
	defer gzipBodyReader.Close()
	fileContent, err := ioutil.ReadAll(gzipBodyReader)
	if err != nil {
		log.Println("[error] gzip.NewReader", err, s.bucket, source)
		return "", err
	}
	return util.ByteSliceToString(fileContent), nil
}

func (s *AWSS3Source) dealer(content string, ops ...func(log *CdnDownloadLogV1)) []CdnDownloadLogV1 {
	lines := strings.Split(content, "\n")
	logs := make([]CdnDownloadLogV1, 0, len(lines))
	nowDate := time.Now().Format("2006-01-02")
	for _, line := range lines {
		lineSplits := strings.Split(line, "\t")
		if len(lineSplits) == 33 {
			day := lineSplits[0]
			ip := lineSplits[4]
			filename := lineSplits[7][1:]
			//source := lineSplits[15]
			if strings.Index(filename, ".tar.gz") != -1 && strings.Index(internalip, ip) == -1 {
				log := CdnDownloadLogV1{Date: day, Component: filename, IP: ip,
					RawData: line, Source: strings.TrimSuffix(s.bucket, "-log"),
					CDN: s.cdn, ResponseTime: int(cast.ToFloat64(lineSplits[18]) * 1000),
					LogTime: day + " " + lineSplits[1], State: lineSplits[13],
					Code: cast.ToInt(lineSplits[8]), Size: cast.ToInt(lineSplits[30]),
					PullDate: nowDate,
				}
				for _, op := range ops {
					op(&log)
				}

				logs = append(logs, log)
			}
		}
	}

	return logs
}

func (s *AWSS3Source) SetAfter(after string) *AWSS3Source {
	s.after = after
	return s
}

func (s *AWSS3Source) Run() {
	if s.after == "" && !s.init {
		source := strings.TrimSuffix(s.bucket, "-log")
		err := s.db.Raw(fmt.Sprintf(`SELECT cdn_download_log_v1.key from cdn_download_log_v1
			where cdn = '%s' and source = '%s' ORDER BY log_time desc limit 1`, s.cdn, source)).
			Scan(&s.after).Error
		if err != nil {
			log.Println("[error] load after err", err, *s)
			return
		}
		log.Println("get after = ", s.after, s.bucket)
	}

	maxRetry := 2
	s3Config := &aws.Config{
		Region:      aws.String(s.region),
		Credentials: credentials.NewStaticCredentials(config.GetString("awsiamid"), config.GetString("awsiamsecret"), ""),
		MaxRetries: &maxRetry,
	}
	s3Session, _ := session.NewSession(s3Config)
	s.client = s3.New(s3Session)

	inputs := s3.ListObjectsV2Input{}
	inputs.SetBucket(s.bucket)
	inputs.SetMaxKeys(100)
	inputs.SetStartAfter(s.after)
	seq := 0
	concurr := util.NewConcurency(15)

s3loop:
	seq++
	output, err := s.client.ListObjectsV2(&inputs)
	if err != nil {
		log.Println("[error] s3Client.ListObjectsV2", err)
		return
	}
	go func() {
		concurr.Run(func(exit func()) {
			defer exit()
			// go
			innerConcurr := util.NewConcurency(10)
			for _, meta := range output.Contents {
				objMeta := meta
				go func() {
					innerConcurr.Run(func(exit func()) {
						defer exit()
						fileContent, err := s.fetcher(*objMeta.Key)
						if err != nil {
							return
						}
						// dealer
						logs := s.dealer(fileContent, func(log *CdnDownloadLogV1) {
							log.Key = *objMeta.Key
						}, ipResolve, version, ipFromGithub)

						Save(logs, s.db)
					})
				}()
			}

			innerConcurr.Wait()
		})
	}()

	if output.NextContinuationToken != nil {
		inputs.SetStartAfter("")
		inputs.SetContinuationToken(*output.NextContinuationToken)
		log.Printf("%d, %d, %#v, %s", seq, len(output.Contents),
			output.Contents[len(output.Contents)-1], s.bucket)
		goto s3loop
	} else {
		s.after = *output.Contents[len(output.Contents)-1].Key
		log.Printf("finish %d, %#v, %s, %s", seq,
			output.Contents[len(output.Contents)-1], awsAfter, s.bucket)
	}

	concurr.Wait()
}

type awssourceCfg struct {
	Bucket string `yaml:"bucket"`
	Region string `yaml:"region"`
	After  string `yaml:"after"`
}

func main() {
	config.InitConfig("config", "yaml", "./etc")
	gdb, err := db.InitDb(config.GetString("db"),
		db.SetMaxIdleConns(config.GetInt("dbidlect")),
		db.SetMaxOpenConns(config.GetInt("dbopenct")))
	if err != nil {
		log.Fatal(err)
	}
	gdb.AutoMigrate(&CdnDownloadLogV1{})

	job.Start()
	ss := []ISource{}
	qiniu := NewQiniuSource(config.GetString("qiniuak"),
		config.GetString("qiniusk"),
		"qiniu",
		config.GetStringSclice("cdnurls"), gdb)
	ss = append(ss, qiniu)
	awscfg := []awssourceCfg{}

	viper.UnmarshalKey("awssources", &awscfg)
	for _, cfg := range awscfg {
		ss = append(ss, newAWSS3Source(cfg.Bucket, "cloudfront", cfg.Region, gdb))
	}

	job.AddJob("30 5 * * ?", job.JobRunner(func() {
		for _, s := range ss {
			s.Run()
		}
	}))

	if config.GetString("runinit") == "true" {
		for _, cfg := range awscfg {
			s3 := newAWSS3Source(cfg.Bucket, "cloudfront", cfg.Region, gdb)
			s3.init = true
			s3.Run()
		}
	}
	dt := os.Getenv("PORT")
	log.Println(dt)

	util.GracefulShutdown(nil, nil)
	log.Println("Server exiting")
}