package main

import (
	"flag"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"code.justin.tv/common/chitin"
	_ "code.justin.tv/common/golibs/bininfo"
	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/pbmsg"
	"code.justin.tv/release/trace/scanproto"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/kinesis"
	"golang.org/x/net/context"
	nettrace "golang.org/x/net/trace"
	"google.golang.org/grpc"
)

const (
	// Region of the Kinesis stream to connect to.
	kinesisRegion = "us-west-2"
)

var (
	_       = flag.String("host", "", "") // TODO: remove once puppet/supervise no longer set this flag
	topic   = flag.String("topic", "trace-transactions", "The Kinesis topic")
	port    = flag.Int("port", 8080, "The port to listen on")
	debug   = flag.String("debug", ":11144", "Debug HTTP port")
	workers = flag.Int("workers", 64, "The number of workers to use for filtering transactions")
)

func init() {
	prev := nettrace.AuthRequest
	nettrace.AuthRequest = func(req *http.Request) (any, sensitive bool) {
		// Allow remote requestors to see what's going on, but don't
		// necessarily give them details.
		any, sensitive = prev(req)
		any = true
		return
	}
}

func main() {
	var wg sync.WaitGroup
	flag.Parse()

	err := chitin.ExperimentalTraceProcessOptIn()
	if err != nil {
		log.Fatalf("trace enable err=%q", err)
	}

	dl, err := net.Listen("tcp", *debug)
	if err != nil {
		log.Fatalf("debug listen err=%q", err)
	}
	go func() {
		err := http.Serve(dl, chitin.Handler(http.DefaultServeMux))
		if err != nil {
			log.Fatalf("debug server err=%q", err)
		}
	}()

	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatalf("failed to listen: %v\n", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	conf := aws.NewConfig().WithRegion(kinesisRegion)
	sess := session.New(conf)
	client := kinesis.New(sess)
	source := scanproto.NewKinesisTransactionSource(ctx, client, *topic)

	manager := newQueryManager()
	server := grpc.NewServer()
	pbmsg.RegisterTraceServer(server, newServer(manager))
	api.RegisterTraceServer(server, newApiServer(manager))

	log.Println("turning on signal handling")
	sigch := make(chan os.Signal, 1)
	signal.Notify(sigch, syscall.SIGINT)
	go func() {
		for ok := true; ok; {
			select {
			case _, ok = <-sigch:
			case _, ok = <-ctx.Done():
			}

			log.Println("server.stop start")
			server.Stop()
			log.Println("server.stop end")

			log.Println("source.stop start")
			source.Stop()
			log.Println("source.stop end")
		}
	}()

	wg.Add(1)
	go func() {
		log.Println("source.run start")
		source.Run()
		log.Println("source.run end")
		wg.Done()
		cancel()
	}()

	wg.Add(1)
	go func() {
		log.Println("source.errors start")
		for e := range source.Errors() {
			log.Printf("source err=%q", e)
		}
		log.Println("source.errors end")
		wg.Done()
	}()

	wg.Add(1)
	go func() {
		log.Println("server.serve start")
		server.Serve(lis)
		log.Println("server.serve end")
		wg.Done()
	}()

	wg.Add(1)
	go func() {
		log.Println("filterQueries start")
		filterQueries(source.Transactions(), manager, *workers)
		log.Println("filterQueries end")
		wg.Done()
	}()

	wg.Wait()
}
