package main

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"net/url"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/golang/protobuf/proto"
	gh "github.com/google/go-github/github"
	"github.com/pkg/errors"
	"github.com/urfave/cli"
	"google.golang.org/grpc"
	pbbar "gopkg.in/cheggaaa/pb.v1"

	"code.justin.tv/dta/rockpaperscissors/client/event"
	"code.justin.tv/dta/rockpaperscissors/client/metric"
	"code.justin.tv/dta/rockpaperscissors/client/projectmetadata"
	internalevent "code.justin.tv/dta/rockpaperscissors/internal/api/event"
	"code.justin.tv/dta/rockpaperscissors/internal/ingestqueueconsumer"
	"code.justin.tv/dta/rockpaperscissors/internal/sandstorm"
	pb "code.justin.tv/dta/rockpaperscissors/proto"
	eventspb "code.justin.tv/dta/rockpaperscissors/proto/events"
)

func init() {
	var eventClient *event.Client
	var projectClient *projectmetadata.Client
	var metricClient *metric.Client
	orgsCmd := cli.Command{
		Name:   "admin",
		Usage:  "special administrative commands",
		Hidden: true,
		Before: func(c *cli.Context) error {
			var err error
			eventClient, err = event.NewClient(c.GlobalString("rps-addr"))
			if err != nil {
				return err
			}
			projectClient, err = projectmetadata.NewClient(c.GlobalString("rps-addr"))
			if err != nil {
				return err
			}
			metricClient, err = metric.NewClient(c.GlobalString("rps-addr"))
			if err != nil {
				return err
			}
			return nil
		},
		After: func(c *cli.Context) error {
			if eventClient != nil {
				if err := eventClient.Close(); err != nil {
					log.Error(err)
				}
			}
			if projectClient != nil {
				if err := projectClient.Close(); err != nil {
					log.Error(err)
				}
			}
			if metricClient != nil {
				if err := metricClient.Close(); err != nil {
					log.Error(err)
				}
			}
			return nil
		},
		Subcommands: []cli.Command{
			{
				Name:  "reattribute-github-pushes",
				Usage: "fill in the attributes on GitHub-push events",
				Action: func(c *cli.Context) error {
					return reAttributeGitHubPushes(eventClient)
				},
			},
			{
				Name:  "ingest-github-stats",
				Usage: "re-ingest commit stats from GitHub-push events",
				Flags: []cli.Flag{
					cli.BoolFlag{
						Name:  "first-clean",
						Usage: "first delete all GitHub-Stats events",
					},
				},
				Action: func(c *cli.Context) error {
					return ingestGitHubStats(eventClient, c.GlobalString("rps-addr"), c.Bool("first-clean"))
				},
			},
			{
				Name:  "get-production-status",
				Usage: "print the latest production incidents event from the past 2 hours",
				Action: func(c *cli.Context) error {
					return getProductionStatus(eventClient)
				},
			},
			{
				Name:      "retype-github-pull-requests",
				Usage:     "change type of GitHub-pull_request events",
				ArgsUsage: "TABLE",
				Flags: []cli.Flag{
					cli.BoolFlag{
						Name:  "enable-writes",
						Usage: "enable writes to the table",
					},
				},
				Action: func(c *cli.Context) error {
					table := c.Args().Get(0)
					if len(table) == 0 {

					}
					return reTypeGitHubPullRequests(table,
						c.Bool("enable-writes"))
				},
			},
			{
				Name:  "benchmark-metrics",
				Usage: "run a sort of benchmark suite to get metric request times",
				Flags: []cli.Flag{
					cli.StringFlag{
						Name:  "team",
						Usage: "fetch all metrics for all projects in this team",
						Value: "DevTools",
					},
				},
				Action: func(c *cli.Context) error {
					return benchmarkMetrics(projectClient, metricClient, c.String("team"))
				},
			},
			{
				Name:      "compress-github-events",
				Usage:     "compress the bodies of GitHub events",
				ArgsUsage: "TABLE",
				Flags: []cli.Flag{
					cli.BoolFlag{
						Name:  "enable-writes",
						Usage: "enable writes to the table",
					},
					cli.StringSliceFlag{
						Name:  "types",
						Usage: "list of the types of events to compress",
						Value: &cli.StringSlice{
							"GitHub-pull_request",
							"GitHub-pull_request-opened",
							"GitHub-pull_request-closed",
							"GitHub-pull_request-review_requested",
							"GitHub-deployment_status",
							"GitHub-ping",
						},
					},
				},
				Action: func(c *cli.Context) error {
					table := c.Args().Get(0)
					if len(table) == 0 {

					}
					return compressGitHubEvents(table,
						c.Bool("enable-writes"), c.StringSlice("types"))
				},
			},
			{
				Name:      "decompress-github-events",
				Usage:     "decompress the bodies of GitHub events",
				ArgsUsage: "TABLE",
				Flags: []cli.Flag{
					cli.BoolFlag{
						Name:  "enable-writes",
						Usage: "enable writes to the table",
					},
					cli.StringSliceFlag{
						Name:  "types",
						Usage: "list of the types of events to decompress",
						Value: &cli.StringSlice{
							"GitHub-pull_request",
							"GitHub-pull_request-opened",
							"GitHub-pull_request-closed",
							"GitHub-pull_request-review_requested",
							"GitHub-deployment_status",
							"GitHub-ping",
						},
					},
				},
				Action: func(c *cli.Context) error {
					table := c.Args().Get(0)
					if len(table) == 0 {

					}
					return deCompressGitHubEvents(table,
						c.Bool("enable-writes"), c.StringSlice("types"))
				},
			},
		},
	}
	app.Commands = append(app.Commands, orgsCmd)
}

func reAttributeGitHubPushes(client *event.Client) error {
	events, err := client.QueryEvents(time.Time{}, time.Now(), "GitHub-push", nil)
	if err != nil {
		return err
	}

	bar := pbbar.StartNew(len(events))
	for _, event := range events {
		err = reAttributeGitHubPush(client, event)
		if err != nil {
			return err
		}
		bar.Increment()
	}
	bar.Finish()

	return nil
}

func reAttributeGitHubPush(client *event.Client, event *pb.Event) error {
	pushEvent := &gh.PushEvent{}
	err := json.Unmarshal(event.Body, pushEvent)
	if err != nil {
		return errors.Wrap(err, "Failed to unmarshal GitHub-push event body")
	}

	if pushEvent.Pusher == nil {
		return errors.Errorf("Event %s has nil pusher",
			base64.StdEncoding.EncodeToString(event.GetUuid()))
	}
	if pushEvent.Pusher.Name == nil {
		return errors.Errorf("Event %s has nil pusher name",
			base64.StdEncoding.EncodeToString(event.GetUuid()))
	}

	event.Attributes = []*pb.Event_Attribute{
		&pb.Event_Attribute{
			Key:   proto.String("github_repository"),
			Value: proto.String(*pushEvent.Repo.URL),
		},
		&pb.Event_Attribute{
			Key:   proto.String("pusher"),
			Value: proto.String(*pushEvent.Pusher.Name),
		},
	}

	err = client.AddEvent(event)
	if err != nil {
		return errors.Wrap(err, "Failed to update event")
	}

	return nil
}

func ingestGitHubStats(eventClient *event.Client, endpoint string, firstClean bool) error {
	awsSession, err := session.NewSessionWithOptions(session.Options{
		SharedConfigState: session.SharedConfigEnable,
	})
	if err != nil {
		return errors.Wrap(err, "Failed to create AWS session")
	}

	sandstormClient := sandstorm.New(awsSession)

	conn, err := grpc.Dial(endpoint, grpc.WithInsecure())
	if err != nil {
		return err
	}
	eventServiceClient := pb.NewEventServiceClient(conn)
	gitHubStatsIngestor := ingestqueueconsumer.NewGitHubStatsIngestor(eventServiceClient, sandstormClient)

	events, err := eventClient.QueryEvents(
		time.Time{}, time.Now(), "GitHub-push", nil)
	if err != nil {
		return err
	}

	bar := pbbar.StartNew(len(events))
	for _, event := range events {
		bar.Increment()

		parsedWebHook, err := gh.ParseWebHook("push", event.Body)
		if err != nil {
			return errors.Wrapf(err,
				"Failed to unmarshal GitHub-push body in event %s",
				base64.StdEncoding.EncodeToString(event.GetUuid()))
		}
		pushEvent, ok := parsedWebHook.(*gh.PushEvent)
		if !ok {
			return errors.Wrapf(err,
				"Parsed webhook in event %s doesn't appear to be a push event.",
				base64.StdEncoding.EncodeToString(event.GetUuid()))
		}

		// We only care about changes on the "default branch".
		defaultBranchRef := fmt.Sprintf(
			"refs/heads/%s", *pushEvent.Repo.DefaultBranch)
		if *pushEvent.Ref == defaultBranchRef {
			continue
		}

		repoURL, err := url.Parse(*pushEvent.Repo.URL)
		if err != nil {
			log.Error(errors.Wrapf(err, "Couldn't parse URL %q", *pushEvent.Repo.URL))
			continue
		}
		repoHost := repoURL.Host
		repoName := *pushEvent.Repo.FullName
		gitHubRepository := &pb.GitHubRepository{
			Host: proto.String(repoHost),
			Name: proto.String(repoName),
		}

		for _, commit := range pushEvent.Commits {
			err = gitHubStatsIngestor.Ingest(context.Background(),
				&pb.IngestGitHubStatsRequest{
					GithubRepository: gitHubRepository,
					CommitSha:        *commit.ID,
				})
			if err != nil {
				log.Error(err)
				continue
			}
		}
	}
	bar.Finish()

	return nil
}

func getProductionStatus(eventClient *event.Client) error {
	end := time.Now()
	start := end.Add(-2 * time.Hour)
	events, err := eventClient.QueryEvents(start, end, "JIRAScraperEvent", nil)
	if err != nil {
		return err
	}

	if len(events) == 0 {
		return errors.New("No JIRAScraperEvents found")
	}

	latestEvent := events[len(events)-1]

	jiraScraperEvent := &eventspb.JIRAScraperEvent{}
	err = proto.Unmarshal(latestEvent.GetBody(), jiraScraperEvent)
	if err != nil {
		return err
	}

	fmt.Print(proto.MarshalTextString(jiraScraperEvent))

	return nil
}

type reTypeStats struct {
	reTyped int
	deleted int
}

func reTypeGitHubPullRequests(table string, enableWrites bool) error {
	awsSession, err := session.NewSessionWithOptions(session.Options{
		SharedConfigState: session.SharedConfigEnable,
	})
	if err != nil {
		return errors.Wrap(err, "Failed to create AWS session")
	}

	stats := &reTypeStats{}
	ctx := context.WithValue(context.Background(), reTypeStats{}, stats)

	// We are using the low-level internal event datastore instead of the
	// gRPC API because delete isn't exposed through the API.
	datastore := internalevent.NewDynamoDBDatastore(awsSession, table)
	datastore.EnableWrites = enableWrites
	results, err := datastore.Query(ctx, &pb.EventDatastoreQuery{
		Type:         "GitHub-pull_request",
		StartSeconds: 0,
		EndSeconds:   time.Now().Unix(),
	})
	if err != nil {
		return err
	}

	events := results.GetEvents()
	bar := pbbar.StartNew(len(events))
	for _, event := range events {
		err = reTypeGitHubPullRequest(ctx, datastore, event)
		if err != nil {
			return err
		}
		bar.Increment()
	}
	bar.Finish()

	fmt.Println("Events re-typed: ", stats.reTyped)
	fmt.Println("Events deleted:  ", stats.deleted)

	return nil
}

func reTypeGitHubPullRequest(ctx context.Context, datastore *internalevent.DynamoDBDatastore, event *pb.Event) error {
	githubEvent, err := gh.ParseWebHook("pull_request", event.GetBody())
	if err != nil {
		return errors.Wrap(err, "Error parsing GitHub event")
	}

	stats := ctx.Value(reTypeStats{}).(*reTypeStats)

	switch githubEvent := githubEvent.(type) {
	case *gh.PullRequestEvent:
		switch githubEvent.GetAction() {
		case "opened":
			fallthrough
		case "closed":
			fallthrough
		case "review_requested":
			event.Type = proto.String(fmt.Sprintf(
				"GitHub-pull_request-%s", githubEvent.GetAction()))
			event.Attributes = []*pb.Event_Attribute{
				{
					Key:   proto.String("head_repository"),
					Value: proto.String(githubEvent.PullRequest.Head.Repo.GetHTMLURL()),
				},
				{
					Key:   proto.String("head_ref"),
					Value: proto.String(githubEvent.PullRequest.Head.GetRef()),
				},
				{
					Key:   proto.String("base_repository"),
					Value: proto.String(githubEvent.PullRequest.Base.Repo.GetHTMLURL()),
				},
				{
					Key:   proto.String("base_ref"),
					Value: proto.String(githubEvent.PullRequest.Base.GetRef()),
				},
				{
					Key:   proto.String("user"),
					Value: proto.String(githubEvent.PullRequest.User.GetLogin()),
				},
			}
			err := datastore.Put(ctx, event)
			if err != nil {
				return err
			}
			stats.reTyped++
		default:
			err := datastore.Delete(ctx, event.GetUuid())
			if err != nil {
				return err
			}
			stats.deleted++
		}
	}
	return nil
}

func benchmarkMetrics(projectClient *projectmetadata.Client, metricClient *metric.Client, teamName string) error {
	projects, err := projectClient.GetTeam(teamName)
	if err != nil {
		return err
	}

	metrics, err := metricClient.ListMetrics()
	if err != nil {
		return err
	}

	numRequests := len(projects) * len(metrics)
	metricPeriodEnd := time.Now()
	metricPeriodStart := metricPeriodEnd.Add(-time.Hour * 7 * 24)

	type benchmarkDataPoint struct {
		metricID  string
		projectID string
		duration  time.Duration
		err       error
	}
	benchmarkDataPoints := make([]*benchmarkDataPoint, numRequests, numRequests)
	sem := make(chan *benchmarkDataPoint, numRequests)

	benchmarkStart := time.Now()
	bar := pbbar.StartNew(numRequests)
	for _, p := range projects {
		for _, m := range metrics {
			go func(metricID string, projectID string) {
				start := time.Now()
				_, err := metricClient.GetMetric(
					metricID, metricPeriodStart, metricPeriodEnd, metric.DAY,
					metric.SelectorOpt{ProjectID: projectID})
				if err != nil {
					log.Error(err)
				}
				sem <- &benchmarkDataPoint{
					metricID:  metricID,
					projectID: projectID,
					duration:  time.Now().Sub(start),
					err:       err,
				}
			}(m.GetMetricId(), p.GetProjectId())
		}
	}
	for i := 0; i < numRequests; i++ {
		dataPoint := <-sem
		benchmarkDataPoints[i] = dataPoint
		bar.Increment()
	}
	bar.Finish()
	benchmarkEnd := time.Now()

	var maxDuration time.Duration
	var longestRequest *benchmarkDataPoint
	var numErrors int
	apdexTolerance := 300 * time.Millisecond
	apdexSatisfiedRequests := 0
	apdexToleratedRequests := 0
	for _, dataPoint := range benchmarkDataPoints {
		if dataPoint.duration < apdexTolerance {
			apdexSatisfiedRequests++
		}
		if dataPoint.duration >= apdexTolerance && dataPoint.duration < apdexTolerance*4 {
			apdexToleratedRequests++
		}
		if dataPoint.duration > maxDuration {
			longestRequest = dataPoint
			maxDuration = dataPoint.duration
		}
		if dataPoint.err != nil {
			numErrors++
		}
	}
	apdexScore := (float32(apdexSatisfiedRequests) + float32(apdexToleratedRequests)/2) / float32(numRequests)

	fmt.Println("Total run time: ", benchmarkEnd.Sub(benchmarkStart))
	fmt.Println("Longest request: ", longestRequest.metricID,
		longestRequest.projectID, longestRequest.duration)
	fmt.Printf("Errors: %0.2f%%\n", float32(numErrors)/float32(numRequests)*100.0)
	fmt.Printf("Apdex Score (%v): %0.2f%%\n", apdexTolerance, apdexScore*100.0)

	return nil
}

type compressStats struct {
	originalBytes int
	newBytes      int
}

func compressGitHubEvents(table string, enableWrites bool, types []string) error {
	awsSession, err := session.NewSessionWithOptions(session.Options{
		SharedConfigState: session.SharedConfigEnable,
	})
	if err != nil {
		return errors.Wrap(err, "Failed to create AWS session")
	}
	datastore := internalevent.NewDynamoDBDatastore(awsSession, table)
	datastore.EnableWrites = enableWrites

	stats := &compressStats{}
	ctx := context.WithValue(context.Background(), compressStats{}, stats)

	events := make([]*pb.Event, 0)
	for _, eventType := range types {
		var results *pb.EventDatastoreQueryResults
		results, err = datastore.Query(ctx, &pb.EventDatastoreQuery{
			Type:         eventType,
			StartSeconds: 0,
			EndSeconds:   time.Now().Unix(),
		})
		if err != nil {
			return err
		}
		fmt.Printf("Found %d %s events\n", len(results.GetEvents()), eventType)
		events = append(events, results.GetEvents()...)
	}

	bar := pbbar.StartNew(len(events))
	for _, event := range events {
		err = compressGitHubEvent(ctx, datastore, event)
		if err != nil {
			return err
		}
		bar.Increment()
	}
	bar.Finish()

	fmt.Println("Bytes saved: ", stats.originalBytes-stats.newBytes)
	fmt.Printf("Compression: %0.2f%%\n", (1.0-float32(stats.newBytes)/float32(stats.originalBytes))*100.0)

	return nil
}

func compressGitHubEvent(ctx context.Context, datastore *internalevent.DynamoDBDatastore, event *pb.Event) error {
	body := event.GetBody()
	stats := ctx.Value(compressStats{}).(*compressStats)

	switch http.DetectContentType(body) {
	case "application/octet-stream", "application/x-gzip":
		return nil
	}

	var b bytes.Buffer
	w := gzip.NewWriter(&b)
	if _, err := w.Write(body); err != nil {
		return err
	}
	if err := w.Close(); err != nil {
		return err
	}

	event.Body = b.Bytes()

	if err := datastore.Put(ctx, event); err != nil {
		return err
	}

	stats.originalBytes += len(body)
	stats.newBytes += len(event.Body)

	return nil
}

func deCompressGitHubEvents(table string, enableWrites bool, types []string) error {
	awsSession, err := session.NewSessionWithOptions(session.Options{
		SharedConfigState: session.SharedConfigEnable,
	})
	if err != nil {
		return errors.Wrap(err, "Failed to create AWS session")
	}
	datastore := internalevent.NewDynamoDBDatastore(awsSession, table)
	datastore.EnableWrites = enableWrites

	ctx := context.Background()

	events := make([]*pb.Event, 0)
	for _, eventType := range types {
		var results *pb.EventDatastoreQueryResults
		results, err = datastore.Query(ctx, &pb.EventDatastoreQuery{
			Type:         eventType,
			StartSeconds: 0,
			EndSeconds:   time.Now().Unix(),
		})
		if err != nil {
			return err
		}
		fmt.Printf("Found %d %s events\n", len(results.GetEvents()), eventType)
		events = append(events, results.GetEvents()...)
	}

	bar := pbbar.StartNew(len(events))
	for _, event := range events {
		err = deCompressGitHubEvent(ctx, datastore, event)
		if err != nil {
			return err
		}
		bar.Increment()
	}
	bar.Finish()

	return nil
}

func deCompressGitHubEvent(ctx context.Context, datastore *internalevent.DynamoDBDatastore, event *pb.Event) error {
	body := event.GetBody()

	switch http.DetectContentType(body) {
	case "application/octet-stream", "application/x-gzip":
		r, err := gzip.NewReader(bytes.NewReader(body))
		if err != nil {
			return err
		}
		body, err = ioutil.ReadAll(r)
		if err != nil {
			return err
		}
		_ = r.Close()

		event.Body = body

		if err := datastore.Put(ctx, event); err != nil {
			return err
		}
	}
	return nil
}
