package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"code.justin.tv/eventbus/client/encryption"

	"github.com/golang/protobuf/ptypes"
	"github.com/pkg/errors"

	"code.justin.tv/eventbus/clock/internal/env"

	"golang.org/x/sync/errgroup"

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

	eventbus "code.justin.tv/eventbus/client"
	"code.justin.tv/eventbus/client/publisher"
	"code.justin.tv/eventbus/client/subscriber/sqsclient"
	"code.justin.tv/eventbus/schema/pkg/clock"

	"code.justin.tv/eventbus/clock/server"
)

const (
	port      = 8000
	debugPort = 6000
	region    = "us-west-2"
)

var targetQueueURLs = map[string]string{
	env.EnvironmentStaging:    "https://sqs.us-west-2.amazonaws.com/065305400592/eventbus-clock-policy_test",
	env.EnvironmentProduction: "https://sqs.us-west-2.amazonaws.com/384110337398/eventbus-7",
}

func main() {
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	done := make(chan struct{})

	awsConfig := aws.NewConfig().WithRegion(region)
	sess, err := session.NewSession(awsConfig)
	if err != nil {
		log.Fatalf("failed to create session: %v", err)
	}

	var publisherEnv publisher.Environment
	if env.Environment() == env.EnvironmentProduction {
		publisherEnv = publisher.EnvProduction
	} else if env.Environment() == env.EnvironmentStaging {
		publisherEnv = publisher.EnvStaging
	} else {
		publisherEnv = publisher.EnvLocal
	}
	log.Printf("Detecting publisher environment %v", publisherEnv)

	publisherConf := publisher.Config{
		Session:     sess,
		Environment: publisherEnv,
		EventTypes:  []string{clock.UpdateEventType},
	}

	clockPublisher, err := publisher.New(publisherConf)
	if err != nil {
		log.Fatalf("failed to create clock publisher: %v", err)
	}

	eventbusSQS, err := subscribeToTicks()
	if err != nil {
		log.Fatalf("could not create eventbus subscriber")
	}

	var wg sync.WaitGroup
	startTwirpServer(done, &wg)

	ctx, cancel := context.WithCancel(context.Background())
	g, ctx := errgroup.WithContext(ctx)

	g.Go(func() error {
		return publishTicks(ctx, *clockPublisher)
	})

	sig := <-signalChan
	log.Printf("Received signal: %v", sig)

	err = eventbusSQS.Shutdown()
	if err != nil {
		log.Printf("could not shutdown eventbus sqs client: %v", err)
	}

	cancel()
	if err := g.Wait(); err != nil {
		log.Fatalf("Unexpected error: %v", err)
	}
}

func publishTicks(ctx context.Context, clockPublisher publisher.Publisher) error {
	defer func() {
		err := clockPublisher.Close()
		if err != nil {
			log.Fatalf("failed to close clock publisher: %v", err)
		}
	}()
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case now := <-ticker.C:
			log.Printf("Publishing ClockUpdate event for time %s...", now)

			ts, err := ptypes.TimestampProto(now)
			if err != nil {
				log.Printf("failed to create timestamp proto: %v", err)
				continue
			}

			tick := &clock.Update{
				Time: ts,
			}

			encryptStart := time.Now()
			err = tick.SetEncryptedSecretTime(&clockPublisher, ptypes.TimestampString(ts))
			if err != nil {
				log.Printf("failed to set ClockUpdate.SecretTime: %v", err)
				continue
			}

			log.Printf("SetEncryptedSecretTime for ClockUpdate event for time %s after %s", now, time.Since(encryptStart))

			// Publishing now happens from a new Clock service running in Lambda on Fulton+CDK
			// See https://code.amazon.com/packages/TwitchClockLambda
			//err = clockPublisher.Publish(ctx, tick)
			//if err != nil {
			//	log.Printf("failed to publish ClockUpdate: %v", err)
			//} else {
			//	log.Printf("Published ClockUpdate event for time %s after %s", now, time.Since(now))
			//}
		}
	}
}

type ClockHandler struct {
	decrypter encryption.Decrypter
}

func (c *ClockHandler) handleUpdate(ctx context.Context, h *eventbus.Header, cu *clock.ClockUpdate) error {
	t := time.Unix(cu.Time.GetSeconds(), int64(cu.Time.GetNanos()))
	log.Printf("Received clock update after %s: %s", time.Since(t), t)

	secretTime, err := cu.GetDecryptedSecretTime(c.decrypter)
	if err != nil {
		return errors.Wrap(err, "could not decrypt SecretTime")
	}
	log.Printf("Decrypted SecretTime: %s", secretTime)

	return nil
}

func subscribeToTicks() (*sqsclient.SQSClient, error) {
	awsConfig := aws.NewConfig().WithRegion(region)
	sess, err := session.NewSession(awsConfig)
	if err != nil {
		log.Printf("could not create new session with region %v: %v", region, err.Error())
		return nil, err
	}

	mux := eventbus.NewMux()
	log.Println("registering ClockUpdateHandler")

	authFieldClient, err := encryption.NewAuthorizedFieldClient(sess)
	if err != nil {
		log.Printf("could not initialize authorized field client: %v", err)
		return nil, err
	}

	clockHandler := &ClockHandler{
		decrypter: authFieldClient,
	}

	clock.RegisterClockUpdateHandler(mux, clockHandler.handleUpdate)

	queueURL, ok := targetQueueURLs[env.Environment()]
	if !ok {
		log.Fatalf("Could not find queue URL config for %s", env.Environment())
	}
	log.Printf("found queue url to subscribe to: %s", queueURL)

	config := sqsclient.Config{
		Session:    sess,
		Dispatcher: mux.Dispatcher(),
		QueueURL:   queueURL,
	}

	return sqsclient.New(config)
}

func startTwirpServer(done <-chan struct{}, wg *sync.WaitGroup) {
	log.Printf("Starting server on port: %d...", port)
	wg.Add(1)
	go func() {
		s := server.New(port, debugPort)
		if err := s.Start(done); err != nil {
			log.Fatal("Failed to start server")
		}
		wg.Done()
	}()
}
