// Package callee implements the s2s auth callee client.
//
// Users should instantiate the Client struct with the required attributes and
// call Start().
package callee

import (
	"errors"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/endpoints"
	"github.com/aws/aws-sdk-go/aws/session"

	identifier "code.justin.tv/amzn/TwitchProcessIdentifier"
	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	cws "code.justin.tv/amzn/TwitchTelemetryCloudWatchMetricsSender"
	"code.justin.tv/sse/malachai/pkg/config"
	"code.justin.tv/sse/malachai/pkg/events"
	"code.justin.tv/sse/malachai/pkg/internal/closer"
	"code.justin.tv/sse/malachai/pkg/internal/inventory"
	"code.justin.tv/sse/malachai/pkg/internal/keystore"
	"code.justin.tv/sse/malachai/pkg/log"
	"code.justin.tv/sse/malachai/pkg/registration"
	"code.justin.tv/sse/malachai/pkg/s2s"
	"code.justin.tv/sse/malachai/pkg/s2s/internal"
	"code.justin.tv/sse/malachai/pkg/signature"
)

const (
	twitchTelemetryCloudwatchFlushPeriod     = 5 * time.Minute
	twitchTelemetryCloudwatchAggregatePeriod = time.Minute
)

type middlewareFunc func(http.Handler) http.Handler

// ClientAPI is a client which provides middleware functions to validate
// authorizations and caller identity.
type ClientAPI interface {
	RequestValidatorMiddleware(inner http.Handler) (fn http.Handler)
	CapabilitiesInjectorMiddleware(inner http.Handler) (fn http.Handler)
	CapabilitiesAuthorizerMiddleware(inner http.Handler) (fn http.Handler)
	Start() error
	Close() error
}

// Client implements ClientAPI.
type Client struct {
	// Required. Name of the registered service.
	ServiceName string

	// Required. EventsLogger writes events to our events pipeline for
	// auditing purposes. Run() and Close() will be called in client.Start()
	// and client.Close() respectively.
	EventsWriterClient events.EventWriterClient

	// optional configuration options
	Config *Config

	// optional user-provided logger
	Logger log.S2SLogger

	eventsLogger events.EventWriter
	events       *eventsClient
	registrar    registration.Registrar
	keyStore     *keystore.Store
	capClient    *calleeCapabilitiesClient
	pim          *pubkeyManager

	sampleObserver telemetry.SampleObserver

	once   sync.Once
	closer closer.API
}

func (h *Client) validateForStart() (err error) {
	if h.ServiceName == "" {
		err = errors.New("ServiceName is required")
		return
	}
	if h.EventsWriterClient == nil {
		err = errors.New("EventsWriterClient is required")
		return
	}
	return
}

// Start initializes the client and starts goroutines. Only does anything
// if the client hasn't been started already or if Close has been called.
func (h *Client) Start() (err error) {
	h.once.Do(func() {
		err = h.validateForStart()
		if err != nil {
			return
		}

		h.Logger = log.NoOpIfNil(h.Logger)

		if h.Config == nil {
			h.Config = &Config{}
		}
		h.Config.calleeName = h.ServiceName
		err = h.Config.FillDefaults()
		if err != nil {
			return
		}

		h.closer = closer.New()

		var sess *session.Session
		if sess, err = session.NewSession(
			config.AWSConfig(h.Config.AWSConfigBase, "", h.Config.roleArn),
		); err != nil {
			return
		}

		var resources *config.Resources
		resources, err = config.GetResources(h.Config.Environment)
		if err != nil {
			return
		}

		cwReporterSession := session.Must(session.NewSession(
			&aws.Config{
				Credentials:         stscreds.NewCredentials(sess, resources.CloudWatchReporterRoleArn),
				Region:              sess.Config.Region,
				STSRegionalEndpoint: endpoints.RegionalSTSEndpoint,
			},
		))
		tPid := &identifier.ProcessIdentifier{
			Service:  "twitch-s2s-callee-firehose",
			Stage:    h.Config.Environment,
			Substage: "primary",
			Region:   aws.StringValue(sess.Config.Region),
		}
		sender := cws.NewUnbufferedWithSession(tPid, nil, cwReporterSession)
		h.sampleObserver = telemetry.NewBufferedAggregator(twitchTelemetryCloudwatchFlushPeriod, 10000, twitchTelemetryCloudwatchAggregatePeriod, sender, nil)
		dimensions := make(telemetry.DimensionSet)
		dimensions["CalleeService"] = h.ServiceName

		reporter := &telemetry.SampleReporter{
			SampleBuilder:  telemetry.SampleBuilder{ProcessIdentifier: *tPid, Dimensions: dimensions},
			SampleObserver: h.sampleObserver,
		}

		if err = h.EventsWriterClient.Init(sess, reporter); err != nil {
			return
		}

		h.eventsLogger = h.EventsWriterClient.
			WithFields(&events.Event{
				CalleeName:      h.ServiceName,
				CalleeID:        h.Config.calleeID,
				PassthroughMode: fmt.Sprintf("%t", h.Config.PassthroughMode),
			})
		go h.EventsWriterClient.Run()

		h.events, err = newEventsClient(h.Config.eventsConfig, h.Logger, h.closer)
		if err != nil {
			return
		}

		err = h.events.start()
		if err != nil {
			return
		}

		// report instance id
		instanceID, err := internal.NewInstanceID()
		if err != nil {
			return
		}

		inventoryClient, err := inventory.New(h.Config.inventoryConfig)
		if err != nil {
			return
		}

		err = inventoryClient.Put(&inventory.Instance{
			ServiceID:   h.Config.calleeID,
			ServiceName: h.ServiceName,
			InstanceID:  instanceID,
			Version:     s2s.Version,
		})
		if err != nil {
			return
		}

		h.keyStore, err = keystore.New(h.Config.KeystoreConfig)
		if err != nil {
			return
		}

		h.pim = newPubkeyManager(h.events.pubkeyEvents, h.keyStore, h.Logger, h.Config.ExpiredPubkeyPurgeInterval, h.closer)
		go h.pim.start()

		h.registrar, err = registration.New(h.Config.RegistrationConfig, h.Logger)
		if err != nil {
			return
		}

		h.capClient, err = newCalleeCapabilitiesClient(h.Config, h.Logger, h.events.capabilities, h.closer)
		if err != nil {
			return
		}
		go h.capClient.start()

	})
	return
}

func chainMiddlewares(inner http.Handler, mws ...middlewareFunc) (handler http.Handler) {
	if len(mws) == 0 {
		return inner
	}
	return mws[0](chainMiddlewares(inner, mws[1:]...))
}

func middlewareHandlerFunc(mws ...middlewareFunc) (handler func(http.Handler) http.Handler) {
	handler = func(inner http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			chainMiddlewares(inner, mws...).ServeHTTP(w, r)
		})
	}
	return
}

// RequestValidatorMiddleware returns a http middleware that injects an
// authorized caller's identity into the request context.
//
// If the caller's identity could not be determined, they will receive a 401.
func (h *Client) RequestValidatorMiddleware(inner http.Handler) (fn http.Handler) {
	mws := h.newRequestValidator()
	middleware := middlewareHandlerFunc(mws...)
	fn = middleware(inner)
	return
}

func (h *Client) newRequestValidator() (mws []middlewareFunc) {

	validator := signature.NewRequestJWTValidator(h.Config.RequestValidatorConfig, h.registrar, h.keyStore, h.Logger, h.Config.StatsReporter)

	rv := newRequestVerifier(h.Config, validator, h.eventsLogger, h.Logger, h.Config.StatsReporter)
	mws = append(mws, h.healthCheckMiddleware, rv)
	return
}

// CapabilitiesInjectorMiddleware returns an http middleware that injects an
// authorized caller's identity and its capabilities into the request context.
//
// If the caller does not have capabilities associated with this callee service,
// they will receive a 403 response.
//
// If the caller's identity could not be determined, they will receive a 401.
func (h *Client) CapabilitiesInjectorMiddleware(inner http.Handler) (fn http.Handler) {
	mws := h.newCapabilitiesInjectorMiddleware()
	middleware := middlewareHandlerFunc(mws...)
	fn = middleware(inner)
	return
}

// CapabilitiesAuthorizerMiddleware returns an http middleware that authorizes
// requests based on simple METHOD PATH matching.
//
// For example, for a caller to be able to call GET /my/path, they will need the
// capability string 'GET /my/path'. Otherwise, they will receive a 403 and the
// wrapped handler will not be called.
//
// If the caller's identity could not be determined, they will receive a 401.
func (h *Client) CapabilitiesAuthorizerMiddleware(inner http.Handler) (fn http.Handler) {
	mws := h.newCapabilitiesInjectorMiddleware()
	mws = append(mws, newCapabilitiesAuthorizer(h.Config, h.eventsLogger, h.Logger))
	middleware := middlewareHandlerFunc(mws...)
	fn = middleware(inner)
	return
}

// returns capablities middleware with authorizer if param withCapAuthorizer is set, middleware with capabilities injector
// is, returned otherwise.
func (h *Client) newCapabilitiesInjectorMiddleware() (mws []middlewareFunc) {
	mws = h.newRequestValidator()
	ci := newCapabilitiesInjector(h.Config, h.capClient, h.eventsLogger, h.Logger)
	mws = append(mws, ci)
	return
}

// healthCheckMiddleware takes the events client and serves a 4XX 5XX errors
// based on the health of the client. for example, sometimes we want to reject
// all requests if we can't update our cache from SQS
func (h *Client) healthCheckMiddleware(inner http.Handler) (fn http.Handler) {
	fn = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		now := time.Now()
		down := h.events.Health().QueueDownSince()
		if !down.IsZero() && now.Sub(down) > h.Config.QueueUnavailableRejectThreshold {
			msg := fmt.Sprintf(
				"request rejected by QueueUnavailablePolicy 'reject': unable to update cache since %s",
				down.String(),
			)
			h.Logger.Warn(msg)
			if h.Config.QueueUnavailablePolicy == QueueUnavailablePolicyReject {
				w.WriteHeader(http.StatusServiceUnavailable)
				_, err := w.Write([]byte(msg))
				if err != nil {
					h.Logger.Errorf("error while writing response: %s", err.Error())
				}
				return
			}
		}
		inner.ServeHTTP(w, r)
	})
	return
}

// Close closes all clients that need to be closed
func (h *Client) Close() (err error) {
	err = h.closer.Close()
	if err != nil {
		h.Logger.Error("unable to stop running goroutines: " + err.Error())
	}

	err = h.EventsWriterClient.Close()
	if err != nil {
		h.Logger.Error("unable to stop events writer client: " + err.Error())
	}

	// stop the metric collector and send all remaining metrics
	h.sampleObserver.Stop()

	// reset Once so Start can be called again
	h.once = sync.Once{}
	return
}
