// Command cored provides the Chain Core daemon and API server.
package chain

import (
	"context"
	"crypto/tls"
	"expvar"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/kr/secureheader"

	"github.com/chainmint/core"
//	"github.com/chainmint/core/accesstoken"
	//"github.com/chainmint/core/blocksigner"
	"github.com/chainmint/core/config"
	//"github.com/chainmint/core/generator"
//	"github.com/chainmint/core/migrate"
	"github.com/chainmint/core/rpc"
	"github.com/chainmint/core/txdb"
	"github.com/chainmint/crypto/ed25519"
	//"github.com/chainmint/database/pg"
	//"github.com/chainmint/database/raft"
	"github.com/chainmint/database/sql"
	"github.com/chainmint/encoding/json"
	"github.com/chainmint/env"
	"github.com/chainmint/errors"
	"github.com/chainmint/generated/rev"
	chainlog "github.com/chainmint/log"
	"github.com/chainmint/log/rotation"
	"github.com/chainmint/log/splunk"
	//"github.com/chainmint/net/http/limit"
	"github.com/chainmint/net/http/reqid"
	"github.com/chainmint/protocol"
	"github.com/chainmint/protocol/bc"
	"github.com/chainmint/protocol/bc/legacy"
	"github.com/chainmint/app"
	"github.com/chainmint/core/generator"
)

const (
	httpReadTimeout  = 2 * time.Minute
	httpWriteTimeout = time.Hour
)

var (
	// config vars
	rootCAs       = env.String("ROOT_CA_CERTS", "") // file path
	listenAddr    = env.String("LISTEN", ":1999")
	//dbURL         = env.String("DATABASE_URL", "postgres:///core?sslmode=disable")
	dbURL         = env.String("DATABASE_URL", "user=gavin password=123456 dbname=core sslmode=disable")
	splunkAddr    = os.Getenv("SPLUNKADDR")
	logFile       = os.Getenv("LOGFILE")
	logSize       = env.Int("LOGSIZE", 5e6) // 5MB
	logCount      = env.Int("LOGCOUNT", 9)
	logQueries    = env.Bool("LOG_QUERIES", false)
	maxDBConns    = env.Int("MAXDBCONNS", 10)           // set to 100 in prod
	rpsToken      = env.Int("RATELIMIT_TOKEN", 0)       // reqs/sec
	rpsRemoteAddr = env.Int("RATELIMIT_REMOTE_ADDR", 0) // reqs/sec
	indexTxs      = env.Bool("INDEX_TRANSACTIONS", true)
	home          = core.HomeDirFromEnvironment()
	bootURL       = env.String("BOOTURL", "")

	// build vars; initialized by the linker
	buildTag    = "?"
	buildCommit = "?"
	buildDate   = "?"

	race []interface{} // initialized in race.go

	// By default, a core is not able to reset its data.
	// This feature can be turned on with the reset build tag.
	//resetIfAllowedAndRequested = func(pg.DB, *raft.Service) {}
)

func init() {
	var version string
	if buildTag != "?" {
		// build tag with chain-core-server- prefix indicates official release
		version = strings.TrimPrefix(buildTag, "chain-core-server-")
	} else {
		// version of the form rev123 indicates non-release build
		version = rev.ID
	}

	expvar.NewString("version").Set(version)
	expvar.NewString("build_tag").Set(buildTag)
	expvar.NewString("build_date").Set(buildDate)
	expvar.NewString("build_commit").Set(buildCommit)
	expvar.NewString("runtime.GOOS").Set(runtime.GOOS)
	expvar.NewString("runtime.GOARCH").Set(runtime.GOARCH)
	expvar.NewString("runtime.Version").Set(runtime.Version())

	config.Version = version
	config.BuildCommit = buildCommit
	config.BuildDate = buildDate
}

func Run(app *app.ChainmintApplication) {
	v := flag.Bool("version", false, "print version information")
	flag.Parse()

	if !*v {
		fmt.Printf("Chain Core starting...\n\n")
	}

	fmt.Printf("cored (Chain Core) %s\n", config.Version)
	fmt.Printf("build-commit: %v\n", config.BuildCommit)
	fmt.Printf("build-date: %v\n", config.BuildDate)
	fmt.Printf("mockhsm: %t\n", config.BuildConfig.MockHSM)
	fmt.Printf("localhost_auth: %t\n", config.BuildConfig.LocalhostAuth)
	fmt.Printf("reset: %t\n", config.BuildConfig.Reset)
	fmt.Printf("http_ok: %t\n", config.BuildConfig.HTTPOk)

	if *v {
		return
	}

	fmt.Printf("\n")

	//maybeMonitorIfOnWindows() // special-case windows

	ctx := context.Background()
	env.Parse()
	//warnCompat(ctx)

	listener, err := net.Listen("tcp", *listenAddr)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	/*
	listener, tlsConfig, err := maybeUseTLS(listener)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}*/

	// TODO(kr): make core.UseTLS take just an http client
	// and use this object in it.
/*	httpClient := new(http.Client)
	httpClient.Transport = &http.Transport{
		TLSClientConfig: tlsConfig,

		// The following fields are default values
		// copied from DefaultTransport.
		// (When you change them, be sure to move them
		// above this line so this comment stays true.)
		DialContext: (&net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
			DualStack: true,
		}).DialContext,
		MaxIdleConns:          100,
		IdleConnTimeout:       90 * time.Second,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
	}

	raftDir := filepath.Join(home, "raft") // TODO(kr): better name for this
	raftDB, err := raft.Start(*listenAddr, raftDir, *bootURL, httpClient, tlsConfig != nil)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	*/

	sql.EnableQueryLogging(*logQueries)
	db, err := sql.Open("hapg", *dbURL)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	db.SetMaxOpenConns(*maxDBConns)
	db.SetMaxIdleConns(*maxDBConns)

/*
	err = migrate.Run(db)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	*/

	//accessTokens := &accesstoken.CredentialStore{DB: db}

	// We add handlers to our serve mux in two phases. In the first phase, we start
	// listening on the raft routes (`/raft`). This allows us to do things like
	// read the config value stored in raft storage. (A new node in a raft cluster
	// can't read values without kicking off a consensus round, which in turn
	// requires this node to be listening for raft requests.)
	//
	// Once this node is able to read the config value, it can set up the remaining
	// cored functionality, and add the rest of the core routes to the serve mux.
	// That is the second phase.
	//
	// The waitHandler accepts incoming requests, but blocks until its underlying
	// handler is set, when the second phase is complete.
	var coreHandler waitHandler
	coreHandler.wg.Add(1)
	mux := http.NewServeMux()
	//mux.Handle("/raft/", raftDB)
	mux.Handle("/", &coreHandler)

	var handler http.Handler = mux
	//handler = core.AuthHandler(handler, raftDB, accessTokens, tlsConfig)
	handler = core.RedirectHandler(handler)
	handler = reqid.Handler(handler)

	secureheader.DefaultConfig.PermitClearLoopback = true
	secureheader.DefaultConfig.HTTPSRedirect = false
	secureheader.DefaultConfig.Next = handler

	server := &http.Server{
		// Note: we should not set TLSConfig here;
		// we took care of TLS with the listener in maybeUseTLS.
		Handler:      secureheader.DefaultConfig,
		ReadTimeout:  httpReadTimeout,
		WriteTimeout: httpWriteTimeout,
		// Disable HTTP/2 for now until the Go implementation is more stable.
		// https://github.com/golang/go/issues/16450
		// https://github.com/golang/go/issues/17071
		TLSNextProto: map[string]func(*http.Server, *tls.Conn, http.Handler){},
	}

	// The `Serve` call has to happen in its own goroutine because
	// it's blocking and we need to proceed to the rest of the core setup after
	// we call it.
	go func() {
		err := server.Serve(listener)
		chainlog.Fatalkv(ctx, chainlog.KeyError, errors.Wrap(err, "Serve"))
	}()
	//resetIfAllowedAndRequested(db, raftDB)

	/*conf, err := config.Load(ctx, db, raftDB)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}*/

	// Initialize internode rpc clients.
	hostname, err := os.Hostname()
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	processID := fmt.Sprintf("chain-%s-%d", hostname, os.Getpid())
	//if conf != nil {
	//	processID += "-" + conf.Id
	//}
	expvar.NewString("processID").Set(processID)

	log.SetPrefix("cored-" + buildTag + ": ")
	log.SetFlags(log.Lshortfile)
	chainlog.SetPrefix(append([]interface{}{"app", "cored", "buildtag", buildTag, "processID", processID}, race...)...)
	chainlog.SetOutput(logWriter())

	conf := "not null"
	var h http.Handler
	var api *core.API
	if &conf != nil {
		api = launchConfiguredCore(ctx, db, processID, core.UseTLS(nil))
	} else {
		var opts []core.RunOption
		//opts = append(opts, core.UseTLS(tlsConfig))
		//opts = append(opts, enableMockHSM(db)...)
		chainlog.Printf(ctx, "Launching as unconfigured Core.")
		api = core.RunUnconfigured(ctx, db, *listenAddr, opts...)
	}
	app.Init(api)
	h = api
	coreHandler.Set(h)
	chainlog.Printf(ctx, "Chain Core online and listening at %s", *listenAddr)

	// block forever without using any resources so this process won't quit while
	// the goroutine containing ListenAndServe is still working
}

// maybeUseTLS loads the TLS cert and key (if so configured)
// and wraps ln in a TLS listener. If using TLS the config
// will be returned. Otherwise the second return arg will
// be nil.
func maybeUseTLS(ln net.Listener) (net.Listener, *tls.Config, error) {
	c, err := core.TLSConfig(
		filepath.Join(home, "tls.crt"),
		filepath.Join(home, "tls.key"),
		*rootCAs,
	)
	if err == core.ErrNoTLS && config.BuildConfig.HTTPOk {
		return ln, nil, nil // files & env vars don't exist; don't want TLS
	} else if err != nil {
		return nil, nil, err
	}
	ln = tls.NewListener(ln, c)
	return ln, c, nil
}

func launchConfiguredCore(ctx context.Context, db *sql.DB, processID string, opts ...core.RunOption) *core.API {
	// Initialize the protocol.Chain.
	heights, err := txdb.ListenBlocks(ctx, *dbURL)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	store := txdb.NewStore(db)
	blockchainId := bc.EmptyStringHash
	c, err := protocol.NewChain(ctx, blockchainId, store, heights)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	// to do: to added BlockSinger.
	gen := generator.New(c, db)
	opts = append(opts, core.GeneratorLocal(gen))

	// Start up the Core. This will start up the various Core subsystems,
	// and begin leader election.
	api, err := core.Run(ctx, db, *dbURL, c, store, *listenAddr, opts...)
	if err != nil {
		chainlog.Fatalkv(ctx, chainlog.KeyError, err)
	}
	return api
}

// remoteHSM is a client wrapper for an hsm that is used as a blocksigner.Signer
type remoteHSM struct {
	Client *rpc.Client
}

func (h *remoteHSM) Sign(ctx context.Context, pk ed25519.PublicKey, bh *legacy.BlockHeader) (signature []byte, err error) {
	body := struct {
		Block *legacy.BlockHeader `json:"block"`
		Pub   json.HexBytes       `json:"pubkey"`
	}{bh, json.HexBytes(pk[:])}
	err = h.Client.Call(ctx, "/sign-block", body, &signature)
	return
}

func remoteSignerInfo(ctx context.Context, processID, buildTag, blockchainID string, conf *config.Config, httpClient *http.Client) (a []*remoteSigner) {
	for _, signer := range conf.Signers {
		u, err := url.Parse(signer.Url)
		if err != nil {
			chainlog.Fatalkv(ctx, chainlog.KeyError, err)
		}
		if len(signer.Pubkey) != ed25519.PublicKeySize {
			chainlog.Fatalkv(ctx, chainlog.KeyError, errors.Wrap(err), "at", "decoding signer public key")
		}
		client := &rpc.Client{
			BaseURL:      u.String(),
			AccessToken:  signer.AccessToken,
			Username:     processID,
			CoreID:       conf.Id,
			BuildTag:     buildTag,
			BlockchainID: blockchainID,
			Client:       httpClient,
		}
		a = append(a, &remoteSigner{Client: client, Key: ed25519.PublicKey(signer.Pubkey)})
	}
	return a
}

// remoteSigner defines the address and public key of another Core
// that may sign blocks produced by this generator.
type remoteSigner struct {
	Client *rpc.Client
	Key    ed25519.PublicKey
}

func (s *remoteSigner) SignBlock(ctx context.Context, marshalledBlock []byte) (signature []byte, err error) {
	err = s.Client.Call(ctx, "/rpc/signer/sign-block", string(marshalledBlock), &signature)
	return
}

func (s *remoteSigner) String() string {
	return s.Client.BaseURL
}

func logWriter() io.Writer {
	dropmsg := []byte("\nlog data dropped\n")
	rotation := &errlog{w: rotation.Create(logFile, *logSize, *logCount)}
	splunk := &errlog{w: splunk.New(splunkAddr, dropmsg)}

	switch {
	case logFile != "" && splunkAddr != "":
		return io.MultiWriter(rotation, splunk)
	case logFile != "" && splunkAddr == "":
		return rotation
	case logFile == "" && splunkAddr != "":
		return splunk
	}
	return os.Stdout
}

type errlog struct {
	w io.Writer
	t time.Time // protected by chain/log mutex
}

func (w *errlog) Write(p []byte) (int, error) {
	// We don't want to ruin our performance
	// when there's a persistent error
	// writing to a log sink.
	// Print to stderr at most once per minute.
	_, err := w.w.Write(p)
	if err != nil && time.Since(w.t) > time.Minute {
		log.Println("chain/log:", err)
		w.t = time.Now()
	}
	return len(p), nil // report success for the MultiWriter
}

type waitHandler struct {
	h  http.Handler
	wg sync.WaitGroup
}

func (wh *waitHandler) Set(h http.Handler) {
	wh.h = h
	wh.wg.Done()
}

func (wh *waitHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	wh.wg.Wait()
	wh.h.ServeHTTP(w, req)
}
