package worker

import (
	"context"
	"sync"
	"time"

	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	"code.justin.tv/creator-collab/log"
	"code.justin.tv/creator-collab/log/errors"
	"code.justin.tv/live/autohost/internal/database/dynamo"
	"code.justin.tv/live/autohost/internal/localdynamo"
	"code.justin.tv/live/autohost/internal/metrics"
	"code.justin.tv/live/autohost/internal/worker/clients/clue"
	"code.justin.tv/live/autohost/internal/worker/clients/multiplex"
	"code.justin.tv/live/autohost/internal/worker/clients/users"
	"code.justin.tv/live/autohost/internal/worker/config"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/endpoints"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
	"github.com/aws/aws-sdk-go/service/ssm"
	"github.com/cactus/go-statsd-client/statsd"
	"golang.org/x/sync/errgroup"
)

const awsRegion = "us-west-2"

// LiveService provides a list of live channels
type LiveService interface {
	LiveChannels(ctx context.Context) (map[string]bool, error)
}

// UsersService provides stream information for a channel
type UsersService interface {
	GetChannelIDs(ctx context.Context, logins []string) (map[string]string, error)
}

// Worker implements the autohost worker algorithm
type Worker struct {
	DynamicConfigLoader *config.DynamicConfigLoader
	Unhoster            *Unhoster
	Server              *Server

	mutex            *sync.Mutex
	errGroup         *errgroup.Group
	cancelTopContext func()
}

// New returns a new Worker struct
func New() (*Worker, error) {
	conf, err := config.GetConfig()
	if err != nil {
		return nil, err
	}
	env := conf.Environment

	awsSession, err := session.NewSession(&aws.Config{
		Region: aws.String(awsRegion),
		// Use the regional STS endpoint when assuming roles instead of the global endpoint.
		// This reduces latency, and prevents our AWS account from getting flagged with an
		// STSGlobalEndpointDeprecation policy engine violation.
		STSRegionalEndpoint: endpoints.RegionalSTSEndpoint,
	})
	if err != nil {
		return nil, errors.Wrap(err, "creating aws session failed")
	}

	secrets, err := config.LoadSecrets(conf, awsSession)
	if err != nil {
		return nil, err
	}

	var logger log.Logger
	if conf.UseDevelopmentLogger {
		logger = log.NewDevelopmentLogger()
	} else {
		logger = log.NewProductionLogger(secrets.RollbarToken, env)
	}

	sampleReporter := newSampleReporter(conf, logger)
	instrumentedAWSSession, err := newInstrumentedAWSSession(sampleReporter)
	if err != nil {
		return nil, err
	}

	statsClient, err := statsd.NewNoopClient()
	if err != nil {
		return nil, errors.Wrap(err, "creating noop statsd client failed")
	}

	dynamoClient, err := newDynamoDBClient(conf, instrumentedAWSSession)
	if err != nil {
		return nil, err
	}
	db := dynamo.New(dynamo.Params{
		Client:            dynamoClient,
		HostTableName:     conf.DynamoDB.HostTableName,
		SettingsTableName: conf.DynamoDB.SettingsTableName,
	})

	clue, err := clue.NewClient(conf.ClientURL.Clue, statsClient, sampleReporter)
	if err != nil {
		return nil, err
	}

	users, err := users.New(conf.ClientURL.Users, statsClient, sampleReporter)
	if err != nil {
		return nil, err
	}

	multiplex := multiplex.NewClient(conf.ClientURL.Multiplex, statsClient, sampleReporter, logger)

	ssmClient := ssm.New(awsSession)
	dynamicConfigLoader := &config.DynamicConfigLoader{
		KeyPrefix: conf.DynamicConfigKeyPrefix,
		SSMClient: ssmClient,
		Logger:    logger,
	}
	server := NewServer(conf.ServerAddress, sampleReporter, logger)

	unhoster := NewUnhoster(UnhosterParams{
		DB:             db,
		Logger:         logger,
		Multiplex:      multiplex,
		SampleReporter: sampleReporter,
		Users:          users,
		Clue:           clue,
	})

	w := &Worker{
		DynamicConfigLoader: dynamicConfigLoader,
		Server:              server,
		Unhoster:            unhoster,

		errGroup: &errgroup.Group{},
		mutex:    &sync.Mutex{},
	}

	return w, nil
}

func newDynamoDBClient(conf *config.Config, awsSession *session.Session) (dynamodbiface.DynamoDBAPI, error) {
	if conf.DynamoDB.LocalDynamo {
		return localdynamo.NewLocalDynamoDBClient(conf.DynamoDB.LocalDynamoEndpoint)
	}

	return dynamodb.New(awsSession), nil
}

func newSampleReporter(conf *config.Config, logger log.Logger) *telemetry.SampleReporter {
	return metrics.NewSampleReporter(&metrics.SampleReporterConfig{
		Environment:             conf.Environment,
		Logger:                  logger,
		Region:                  awsRegion,
		RunningInEC2:            conf.RunningInEC2,
		SendMetricsToCloudwatch: conf.EnableCloudWatchMetrics,
		ServiceName:             "Autohost Worker",
	})
}

func newInstrumentedAWSSession(sampleReporter *telemetry.SampleReporter) (*session.Session, error) {
	awsSession, err := session.NewSession(&aws.Config{
		Region: aws.String(awsRegion),
		// Use the regional STS endpoint when assuming roles instead of the global endpoint.
		// This reduces latency, and prevents our AWS account from getting flagged with an
		// STSGlobalEndpointDeprecation policy engine violation.
		STSRegionalEndpoint: endpoints.RegionalSTSEndpoint,
	})
	if err != nil {
		return nil, errors.Wrap(err, "creating aws session failed")
	}

	// Add middleware to the AWS session to report metrics to Cloudwatch.
	metricClient := metrics.NewAWSMiddleware(sampleReporter)
	awsSession = metricClient.AddToSession(awsSession)

	return awsSession, nil
}

// Start starts go routines for each component.
func (w *Worker) Start() {
	ctx, cancel := context.WithCancel(context.Background())
	w.mutex.Lock()
	w.cancelTopContext = cancel
	w.mutex.Unlock()

	// Start the HTTP server with the health check.
	w.errGroup.Go(w.Server.Start)

	// Start the loop that polls for live channels, and figures out when channels have gone live
	// or offline.
	w.errGroup.Go(func() error {
		w.Unhoster.Process(ctx)
		return nil
	})
}

// Stop signals each component to start shutting down.
func (w *Worker) Stop() {
	// Start shutting down the HTTP server.
	w.errGroup.Go(func() error {
		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
		w.Server.Close(ctx)
		cancel()

		return nil
	})

	// Signal context aware goroutines to exit.
	var cancelTopContext func()
	w.mutex.Lock()
	cancelTopContext = w.cancelTopContext
	w.mutex.Unlock()

	if cancelTopContext != nil {
		cancelTopContext()
	}
}

// Wait blocks until all the main go-routines have exited.
// It returns the first non-nil error returned by a main go-routine, or nil if there were
// no errors.
func (w *Worker) Wait() error {
	return w.errGroup.Wait()
}

func isDone(ctx context.Context) bool {
	select {
	case <-ctx.Done():
		return true
	default:
		return false
	}
}
