// LoadSchemaJSON loads the formal representation of a schema
// from a JSON file and returns the more usable denormalized
// representaion (Schema) for it.
func LoadSchemaJSON(filename string) (schema *Schema, err error) {
	var source SchemaFormal
	if err := jscfg.ReadJson(filename, &source); err != nil {
		return nil, err
	}
	return BuildSchema(&source)
}

func Init(socketFile, dbCredentialsFile string) (DBConfigs, error) {
	var err error
	if dbCredentialsFile != "" {
		dbCreds := make(dbCredentials)
		if err = jscfg.ReadJson(dbCredentialsFile, &dbCreds); err != nil {
			return dbConfigs, err
		}
		if passwd, ok := dbCreds[dbConfigs.App.Uname]; ok {
			dbConfigs.App.Pass = passwd[0]
		}
		if passwd, ok := dbCreds[dbConfigs.Dba.Uname]; ok {
			dbConfigs.Dba.Pass = passwd[0]
		}
		if passwd, ok := dbCreds[dbConfigs.Repl.Uname]; ok {
			dbConfigs.Repl.Pass = passwd[0]
		}
	}
	if socketFile != "" {
		dbConfigs.App.UnixSocket = socketFile
		dbConfigs.Dba.UnixSocket = socketFile
		dbConfigs.Repl.UnixSocket = socketFile
	}
	log.Infof("DBConfigs: %s\n", dbConfigs.Redacted())
	return dbConfigs, err
}

func loadSchemaOverrides(overridesFile string) []ts.SchemaOverride {
	var schemaOverrides []ts.SchemaOverride
	if err := jscfg.ReadJson(overridesFile, &schemaOverrides); err != nil {
		log.Warningf("can't read overridesFile %v: %v", overridesFile, err)
	} else {
		data, _ := json.MarshalIndent(schemaOverrides, "", "  ")
		log.Infof("schemaOverrides: %s\n", data)
	}
	return schemaOverrides
}

func Init(socketFile, dbConfigsFile, dbCredentialsFile string) (dbcfgs DBConfigs, err error) {
	dbcfgs.App = DBConfig{
		Uname:   "vt_app",
		Charset: "utf8",
	}
	dbcfgs.Dba = mysql.ConnectionParams{
		Uname:   "vt_dba",
		Charset: "utf8",
	}
	dbcfgs.Repl = mysql.ConnectionParams{
		Uname:   "vt_repl",
		Charset: "utf8",
	}
	if dbConfigsFile != "" {
		if err = jscfg.ReadJson(dbConfigsFile, &dbcfgs); err != nil {
			return
		}
	}

	if dbCredentialsFile != "" {
		dbCreds := make(dbCredentials)
		if err = jscfg.ReadJson(dbCredentialsFile, &dbCreds); err != nil {
			return
		}
		if passwd, ok := dbCreds[dbcfgs.App.Uname]; ok {
			dbcfgs.App.Pass = passwd[0]
		}
		if passwd, ok := dbCreds[dbcfgs.Dba.Uname]; ok {
			dbcfgs.Dba.Pass = passwd[0]
		}
		if passwd, ok := dbCreds[dbcfgs.Repl.Uname]; ok {
			dbcfgs.Repl.Pass = passwd[0]
		}
	}
	dbcfgs.App.UnixSocket = socketFile
	dbcfgs.Dba.UnixSocket = socketFile
	dbcfgs.Repl.UnixSocket = socketFile
	log.Infof("%s: %s\n", dbConfigsFile, dbcfgs.Redacted())
	return
}

// InitQueryService registers the query service, after loading any
// necessary config files. It also starts any relevant streaming logs.
func InitQueryService() {
	SqlQueryLogger.ServeLogs(*queryLogHandler)
	TxLogger.ServeLogs(*txLogHandler)

	qsConfig := DefaultQsConfig
	if *qsConfigFile != "" {
		if err := jscfg.ReadJson(*qsConfigFile, &qsConfig); err != nil {
			log.Fatalf("cannot load qsconfig file: %v", err)
		}
	}

	RegisterQueryService(qsConfig)
}

// InitQueryService registers the query service, after loading any
// necessary config files. It also starts any relevant streaming logs.
func InitQueryService() {
	SqlQueryLogger.ServeLogs(*queryLogHandler)
	TxLogger.ServeLogs(*txLogHandler)

	qsConfig := DefaultQsConfig
	if *qsConfigFile != "" {
		if err := jscfg.ReadJson(*qsConfigFile, &qsConfig); err != nil {
			log.Fatalf("cannot load qsconfig file: %v", err)
		}
	}
	qsConfig.SpotCheckRatio = *spotCheckRatio
	// TODO(liguo): Merge into your CL
	qsConfig.StreamExecThrottle = *streamExecThrottle
	qsConfig.StreamWaitTimeout = *streamWaitTimeout

	RegisterQueryService(qsConfig)
}

func (fcs *FileCredentialsServer) GetUserAndPassword(user string) (string, string, error) {
	fcs.mu.Lock()
	defer fcs.mu.Unlock()

	if *dbCredentialsFile == "" {
		return "", "", ErrUnknownUser
	}

	// read the json file only once
	if fcs.dbCredentials == nil {
		fcs.dbCredentials = make(map[string][]string)
		if err := jscfg.ReadJson(*dbCredentialsFile, &fcs.dbCredentials); err != nil {
			log.Warningf("Failed to read dbCredentials file: %v", *dbCredentialsFile)
			return "", "", err
		}
	}

	if passwd, ok := fcs.dbCredentials[user]; !ok {
		return "", "", ErrUnknownUser
	} else {
		return user, passwd[0], nil
	}
}

func initQueryService(dbcfgs dbconfigs.DBConfigs) {
	ts.SqlQueryLogger.ServeLogs("/debug/querylog")
	ts.TxLogger.ServeLogs("/debug/txlog")

	if err := jscfg.ReadJson(*qsConfigFile, &qsConfig); err != nil {
		relog.Warning("%s", err)
	}
	ts.RegisterQueryService(qsConfig)
	usefulLameDuckPeriod := float64(qsConfig.QueryTimeout + 1)
	if usefulLameDuckPeriod > *lameDuckPeriod {
		*lameDuckPeriod = usefulLameDuckPeriod
		relog.Info("readjusted -lame-duck-period to %f", *lameDuckPeriod)
	}
	if *queryLog != "" {
		if f, err := os.OpenFile(*queryLog, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644); err == nil {
			ts.QueryLogger = relog.New(f, "", relog.DEBUG)
		} else {
			relog.Fatal("Error opening file %v: %v", *queryLog, err)
		}
	}
	umgmt.AddCloseCallback(func() {
		ts.DisallowQueries(true)
	})
}

func main() {
	dbConfigsFile, dbCredentialsFile := dbconfigs.RegisterCommonFlags()
	flag.Parse()

	if err := servenv.Init("vttablet"); err != nil {
		relog.Fatal("Error in servenv.Init: %s", err)
	}

	tabletAlias := tabletParamToTabletAlias(*tabletPath)

	mycnf := readMycnf(tabletAlias.Uid)
	dbcfgs, err := dbconfigs.Init(mycnf.SocketFile, *dbConfigsFile, *dbCredentialsFile)
	if err != nil {
		relog.Warning("%s", err)
	}
	dbcfgs.App.Memcache = *rowcache

	if err := jscfg.ReadJson(*overridesFile, &schemaOverrides); err != nil {
		relog.Warning("%s", err)
	} else {
		data, _ := json.MarshalIndent(schemaOverrides, "", "  ")
		relog.Info("schemaOverrides: %s\n", data)
	}

	initQueryService(dbcfgs)
	initUpdateStreamService(mycnf)
	ts.RegisterCacheInvalidator()                                                   // depends on both query and updateStream
	err = initAgent(tabletAlias, dbcfgs, mycnf, *dbConfigsFile, *dbCredentialsFile) // depends on both query and updateStream
	if err != nil {
		relog.Fatal("%s", err)
	}

	rpc.HandleHTTP()

	// NOTE(szopa): Changing credentials requires a server
	// restart.
	if *authConfig != "" {
		if err := auth.LoadCredentials(*authConfig); err != nil {
			relog.Error("could not load authentication credentials, not starting rpc servers: %v", err)
		}
		serveAuthRPC()
	}

	serveRPC()

	// make a list of paths we can serve HTTP traffic from.
	// we don't resolve them here to real paths, as they might not exits yet
	snapshotDir := mysqlctl.SnapshotDir(tabletAlias.Uid)
	allowedPaths := []string{
		path.Join(vtenv.VtDataRoot(), "data"),
		mysqlctl.TabletDir(tabletAlias.Uid),
		snapshotDir,
		mycnf.DataDir,
		mycnf.InnodbDataHomeDir,
		mycnf.InnodbLogGroupHomeDir,
	}

	// NOTE: trailing slash in pattern means we handle all paths with this prefix
	http.Handle(mysqlctl.SnapshotURLPath+"/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		handleSnapshot(w, r, snapshotDir, allowedPaths)
	}))

	// we delegate out startup to the micromanagement server so these actions
	// will occur after we have obtained our socket.
	umgmt.SetLameDuckPeriod(float32(*lameDuckPeriod))
	umgmt.SetRebindDelay(float32(*rebindDelay))
	umgmt.AddStartupCallback(func() {
		umgmt.StartHttpServer(fmt.Sprintf(":%v", *port))
		if *securePort != 0 {
			relog.Info("listening on secure port %v", *securePort)
			umgmt.StartHttpsServer(fmt.Sprintf(":%v", *securePort), *cert, *key, *caCert)
		}
	})
	umgmt.AddStartupCallback(func() {
		c := make(chan os.Signal, 1)
		signal.Notify(c, syscall.SIGTERM)
		go func() {
			for sig := range c {
				umgmt.SigTermHandler(sig)
			}
		}()
	})

	relog.Info("started vttablet %v", *port)
	umgmtSocket := fmt.Sprintf("/tmp/vttablet-%08x-umgmt.sock", *port)
	if umgmtErr := umgmt.ListenAndServe(umgmtSocket); umgmtErr != nil {
		relog.Error("umgmt.ListenAndServe err: %v", umgmtErr)
	}
	relog.Info("done")
}