// Init is called to initialize the application.
func Init(configKey string) {

	// Init the configuration system.
	if err := cfg.Init(cfg.EnvProvider{Namespace: configKey}); err != nil {
		fmt.Println("Error initalizing configuration system", err)
		os.Exit(1)
	}

	// Init the log system.
	logLevel := func() int {
		ll, err := cfg.Int(cfgLoggingLevel)
		if err != nil {
			return log.USER
		}
		return ll
	}
	log.Init(os.Stderr, logLevel)

	// Log all the configuration options
	log.User("startup", "Init", "\n\nConfig Settings: %s\n%s\n", configKey, cfg.Log())

	// Load user defined custom headers. HEADERS should be key:value,key:value
	if hs, err := cfg.String("HEADERS"); err == nil {
		hdrs := strings.Split(hs, ",")
		for _, hdr := range hdrs {
			if kv := strings.Split(hdr, ":"); len(kv) == 2 {
				log.User("startup", "Init", "User Headers : %s:%s", kv[0], kv[1])
				app.userHeaders[kv[0]] = kv[1]
			}
		}
	}
}

// Import gets data from report on failed import, transform it and send it to pillar
func importOnlyFailedRecords() { //dbsource source.Sourcer, limit int, offset int, orderby string, thisStrategy string, reportOnFailedRecords bool) {

	log.User(uuid, "sponge.importOnlyFailedRecords", "### Reading file of data to import.")

	// get the data that needs to be imported
	tables, err := report.ReadReport(options.Reportdbfile) //map[model]map[id]interface{}
	if err != nil {
		log.Error(uuid, "sponge.importOnlyFailedRecords", err, "Getting the rows that will be imported")
	}

	var data []map[string]interface{}

	for table, ids := range tables {

		if len(ids) < 1 { // only one ID
			// Get the data
			log.User(uuid, "sponge.importOnlyFailedRecords", "### Reading data for entity '%s'. \n", table)
			data, err = dbsource.GetData(table, &options) //options.offset, options.limit, options.orderby, options.query)
		} else {
			log.User(uuid, "sponge.importOnlyFailedRecords", "### Reading data for entity '%s', quering '%s'. \n", table, ids)
			data, err = dbsource.GetQueryData(table, &options, ids)
		}
		if err != nil && options.ReportOnFailedRecords {
			report.Record(table, ids, "Failing getting data", err)
		}

		// transform and get data into pillar
		process(table, data)
	}
}

func importFromDB(collections []string) {
	// var data []map[string]interface{}

	for _, name := range collections { // Reads through all the collections whose transformations are in the strategy configuration file

		foreignEntity := source.GetForeignEntity(name)

		log.User(uuid, "sponge.importAll", "### Reading data to import from %s into collection '%s'. \n", foreignEntity, name)

		// Get the data
		data, err := dbsource.GetData(name, &options) //options.offset, options.limit, options.orderby, "")
		if err != nil {
			log.Error(uuid, "sponge.importAll", err, "Get external data for collection %s.", name)
			//RECORD to report about failing modelName
			if options.ReportOnFailedRecords {
				report.Record(name, "", "Failing to get data.", err)
			}
			continue
		}

		log.User(uuid, "sponge.importAll", "### Transforming data and sending it to Coral. \n")
		//transform and send to pillar the data
		process(name, data)
	}
}

func process(coralName string, data []map[string]interface{}) {

	// Transform the data row by row
	log.User(uuid, "sponge.process", "# Transforming data to the coral schema.\n")
	log.User(uuid, "sponge.process", "# And importing %v documents.", len(data))

	// Initialize benchmarking for current table
	start := time.Now()
	blockStart := time.Now()
	blockSize := int64(1000) // number of documents between each report
	documents := int64(0)
	totalDocuments := int64(len(data))

	for _, row := range data {

		// output benchmarking for each block of documents
		if documents%blockSize == 0 && documents > 0 {

			// calculate stats
			percentComplete := float64(documents) / float64(totalDocuments) * float64(100)
			msSinceStart := time.Since(start).Nanoseconds() / int64(1000000)
			msSinceBlock := time.Since(blockStart).Nanoseconds() / int64(1000000)
			timeRemaining := int64(float64(time.Since(start).Seconds()) / float64(percentComplete) * float64(100))

			//log stats
			log.User(uuid, "sponge.process", "%v%% (%v/%v imported) %vms, %vms avg - last %v in %vms, %vms avg -- est time remaining %vs\n", int64(percentComplete), documents, totalDocuments, msSinceStart, msSinceStart/documents, blockSize, msSinceBlock, msSinceBlock/blockSize, int64(timeRemaining))
			blockStart = time.Now()

		}
		documents = documents + 1

		// transform the row
		id, newRows, err := fiddler.TransformRow(row, coralName)
		if err != nil {
			log.Error(uuid, "sponge.process", err, "Error when transforming the row %s.", row)
			//RECORD to report about failing transformation
			if options.ReportOnFailedRecords {
				report.Record(coralName, id, "Failing transform data", err)
			}
		}

		// Usually newRows only will have a document but in the case that we have subcollections
		// we may get more than one document from a transformation
		for _, newRow := range newRows {

			log.Dev(uuid, "sponge.process", "Transforming: %v into %v.", row, newRow)

			// send the row to pillar
			err = coral.AddRow(newRow, coralName)
			if err != nil {
				log.Error(uuid, "sponge.process", err, "Error when adding a row") // thae row %v to %s.", string(newRow), modelName)
				//RECORD to report about failing adding row to coral db
				if options.ReportOnFailedRecords {
					report.Record(coralName, id, "Failing add row to coral", err)
				}
			}
		}
	}
}

// Job implements the Jobber interface so task can be managed.
func (t *Task) Job(context interface{}) error {
	log.User(context, "Job", "Started : **********")

	time.Sleep(time.Second)

	log.User(context, "Job", "Completed : **********")
	return nil
}

// Run performs the execution of the specified job.
func Run(context interface{}, timeout time.Duration, job Jobber) error {
	log.User(context, "Run", "Started : Timeout[%v]", timeout)

	// Init the runner for use.
	if initRunner(timeout) {
		return errors.New("Already running")
	}

	// When the task is done reset everything.
	defer resetRunner()

	// We want to receive all interrupt based signals.
	signal.Notify(runner.sigChan, os.Interrupt)

	// Launch the processor.
	go processor(context, job)

	for {
		select {
		case <-runner.sigChan:

			// Interrupt event signaled by the operation system.
			log.User(context, "Run", "Interrupt Received")

			// Flag we received the signal.
			runner.recvShutdown = true

			// Close the channel to signal to the processor
			// it needs to shutdown.
			close(runner.shutdown)

			// No need to process anymore events.
			signal.Stop(runner.sigChan)

		case <-runner.kill:

			// We have taken too much time. Kill the app hard.
			log.User(context, "Run", "Completed : Timedout")
			return ErrTimeout

		case err := <-runner.complete:

			// Everything completed within the time given.
			log.User(context, "Run", "Completed : Task Result : %v", err)

			if runner.recvShutdown {
				return ErrSignaled
			}

			return err
		}
	}
}

// Handle is our mechanism for mounting Handlers for a given HTTP verb and path
// pair, this makes for really easy, convenient routing.
func (a *App) Handle(verb, path string, handler Handler, mw ...Middleware) {

	// The function to execute for each request.
	h := func(w http.ResponseWriter, r *http.Request, p map[string]string) {
		start := time.Now()

		var dbConn *db.DB
		if app.useMongo {
			dbConn = db.NewMGO()
		}

		c := Context{
			DB:             dbConn,
			ResponseWriter: w,
			Request:        r,
			Params:         p,
			SessionID:      uuid.New(),
		}

		if app.useMongo {
			defer c.DB.CloseMGO()
		}

		log.User(c.SessionID, "Request", "Started : Method[%s] URL[%s] RADDR[%s]", c.Request.Method, c.Request.URL.Path, c.Request.RemoteAddr)

		// Wrap the handler in all associated middleware.
		wrap := func(h Handler) Handler {
			// Wrap up the application-wide first...
			for i := len(a.mw) - 1; i >= 0; i-- {
				h = a.mw[i](h)
			}

			// Then wrap with our route specific ones.
			for i := len(mw) - 1; i >= 0; i-- {
				h = mw[i](h)
			}

			return h
		}

		// Call the wrapped handler and handle any possible error.
		if err := wrap(handler)(&c); err != nil {
			c.Error(err)
		}

		log.User(c.SessionID, "Request", "Completed : Status[%d] Duration[%s]", c.Status, time.Since(start))
	}

	// Add this handler for the specified verb and route.
	a.TreeMux.Handle(verb, path, h)
}

// Init is called to initialize the application.
func Init(configKey string) {

	// Init the configuration system.
	if err := cfg.Init(cfg.EnvProvider{Namespace: configKey}); err != nil {
		fmt.Println("Error initalizing configuration system", err)
		os.Exit(1)
	}

	// Init the log system.
	logLevel := func() int {
		ll, err := cfg.Int(cfgLoggingLevel)
		if err != nil {
			return log.USER
		}
		return ll
	}
	log.Init(os.Stderr, logLevel)

	// Log all the configuration options
	log.User("startup", "Init", "\n\nConfig Settings: %s\n%s\n", configKey, cfg.Log())

	// Init MongoDB if configured.
	if _, err := cfg.String(cfgMongoHost); err == nil {
		app.useMongo = true

		cfg := mongo.Config{
			Host:     cfg.MustString(cfgMongoHost),
			AuthDB:   cfg.MustString(cfgMongoAuthDB),
			DB:       cfg.MustString(cfgMongoDB),
			User:     cfg.MustString(cfgMongoUser),
			Password: cfg.MustString(cfgMongoPassword),
		}

		if err := mongo.Init(cfg); err != nil {
			log.Error("startup", "Init", err, "Initializing MongoDB")
			os.Exit(1)
		}
	}

	// Load user defined custom headers. HEADERS should be key:value,key:value
	if hs, err := cfg.String("HEADERS"); err == nil {
		hdrs := strings.Split(hs, ",")
		for _, hdr := range hdrs {
			if kv := strings.Split(hdr, ":"); len(kv) == 2 {
				log.User("startup", "Init", "User Headers : %s:%s", kv[0], kv[1])
				app.userHeaders[kv[0]] = kv[1]
			}
		}
	}
}

func main() {
	const context = "startup"

	// Create the configuration.
	cfg := udp.Config{
		NetType: "udp4",
		Addr:    ":6000",

		ConnHandler: udpConnHandler{},
		ReqHandler:  udpReqHandler{},
		RespHandler: udpRespHandler{},

		OptIntPool: udp.OptIntPool{
			RecvMinPoolSize: func() int { return 2 },
			RecvMaxPoolSize: func() int { return 100 },
			SendMinPoolSize: func() int { return 2 },
			SendMaxPoolSize: func() int { return 100 },
		},
	}

	// Create a new UDP value.
	u, err := udp.New(context, "Sample", cfg)
	if err != nil {
		log.Error(context, "main", err, "Creating udp")
		return
	}

	// Start accepting client data.
	if err := u.Start(context); err != nil {
		log.Error(context, "main", err, "Starting udp")
		return
	}

	// Defer the stop on shutdown.
	defer u.Stop(context)

	log.User(context, "main", "Waiting for data on: %s", u.Addr())

	// Listen for an interrupt signal from the OS.
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt)
	<-sigChan

	// Use netcat to test the server.
	// nc -4u localhost 6000 < test.hex

	log.User(context, "main", "Shutting down")
}

// GetData returns the raw data from the external source, entityname entity
func (m MongoDB) GetData(entityname string, options *Options) ([]map[string]interface{}, error) { // offset int, limit int, orderby string, query string) ([]map[string]interface{}, bool, error) { //(*sql.Rows, error) {

	var data []map[string]interface{}

	//fields := strategy.GetEntityForeignFields(entityname) //[]]map[string]string

	// open a connection
	session, err := m.initSession()
	if err != nil {
		log.Error(uuid, "mongodb.getdata", err, "Initializing mongo session.")
		return nil, err
	}
	defer m.closeSession(session)

	credentialD, ok := credential.(str.CredentialDatabase)
	if !ok {
		err = fmt.Errorf("Error asserting type CredentialDatabase from interface Credential.")
		log.Error(uuid, "mongodb.getdata", err, "Asserting Type CredentialDatabase")
		return nil, err
	}
	cred := mgo.Credential{
		Username: credentialD.Username,
		Password: credentialD.Password,
	}

	err = session.Login(&cred)
	if err != nil {
		log.Error(uuid, "mongodb.getdata", err, "Login mongo session.")
		return nil, err
	}

	db := session.DB(credentialD.Database)
	col := db.C(entityname)

	var mquery map[string]interface{}
	if options.Query != "" {
		err = json.Unmarshal([]byte(options.Query), &mquery)
		if err != nil {
			log.Error(uuid, "mongodb.getdata", err, "Unmarshalling query %v", options.Query)
			return nil, err
		}
	}

	//.Select(fieldsToGet) <--- I'm not using Select because SOME FIELDS IN THE TRANSLATION FILE ARE NOT THE RIGHT ONES TO DO THE SELECT. For example: context.object.0.uri
	err = col.Find(mquery).Skip(options.Offset).Limit(options.Limit).All(&data)
	if err != nil {
		log.Error(uuid, "mongodb.getdata", err, "Getting collection %s.", entityname)
		return nil, err
	}

	log.User(uuid, "mongodb.getdata", "### Flattening documents to be able to translate them. \n")

	flattenData, err := flattenizeData(data)
	if err != nil {
		log.Error(uuid, "mongodb.getdata", err, "Normalizing data from mongo to fit into fiddler.")
		return nil, err
	}

	return flattenData, nil
}

func main() {
	const context = "main"

	// Create a task value for execution.
	t := Task{
		Name: "test task",
	}

	// Start the job running with a specified duration.
	if err := runner.Run(context, time.Second, &t); err != nil {
		switch err {
		case runner.ErrTimeout:

			// The task did not finish within the specified duration.
			log.Error(context, "main", err, "Task timeout")

		case runner.ErrSignaled:

			// The user hit <control> c and we shutdown early.
			log.Error(context, "main", err, "Shutdown early")

		default:

			// An error occurred in the processing of the task.
			log.Error(context, "main", err, "Processing error")
		}

		os.Exit(1)
	}

	log.User(context, "main", "Completed")
}

func init() {

	// This is being added to showcase configuration.
	os.Setenv("KIT_LOGGING_LEVEL", "1")
	os.Setenv("KIT_MIN_ROUTINES", "1")
	os.Setenv("KIT_MAX_ROUTINES", "10")

	// Init the configuration system.
	if err := cfg.Init(cfg.EnvProvider{Namespace: configKey}); err != nil {
		fmt.Println("Error initalizing configuration system", err)
		os.Exit(1)
	}

	// Init the log system.
	logLevel := func() int {
		ll, err := cfg.Int(cfgLoggingLevel)
		if err != nil {
			return log.USER
		}
		return ll
	}
	log.Init(os.Stderr, logLevel)

	// Log all the configuration options
	log.User("startup", "init", "\n\nConfig Settings: %s\n%s\n", configKey, cfg.Log())
}

// TestLogLevelDEV tests the basic functioning of the logger in DEV mode.
func TestLogLevelDEV(t *testing.T) {
	t.Log("Given the need to log DEV and USER messages.")
	{
		t.Log("\tWhen we set the logging level to DEV.")
		{
			log.Init(&logdest, func() int { return log.DEV })
			resetLog()
			defer displayLog()

			dt := time.Now().Format("2006/01/02 15:04:05")

			log1 := fmt.Sprintf("%s log_test.go:81: DEV : context : FuncName : Message 1 no format\n", dt)
			log2 := fmt.Sprintf("%s log_test.go:82: USER : context : FuncName : Message 2 with format: A, B\n", dt)
			log3 := fmt.Sprintf("%s log_test.go:83: ERROR : context : FuncName : An error : Message 3 with format: C, D\n", dt)

			log.Dev("context", "FuncName", "Message 1 no format")
			log.User("context", "FuncName", "Message 2 with format: %s, %s", "A", "B")
			log.Error("context", "FuncName", errors.New("An error"), "Message 3 with format: %s, %s", "C", "D")

			if logdest.String() == log1+log2+log3 {
				t.Logf("\t\t%v : Should log the expected trace line.", succeed)
			} else {
				t.Log("***>", logdest.String())
				t.Log("***>", log1+log2+log3)
				t.Errorf("\t\t%v : Should log the expected trace line.", failed)
			}
		}
	}
}

// Respond sends JSON to the client.
// If code is StatusNoContent, v is expected to be nil.
func (c *Context) Respond(data interface{}, code int) {
	log.User(c.SessionID, "api : Respond", "Started : Code[%d]", code)

	c.Status = code

	if code == http.StatusNoContent {
		c.WriteHeader(http.StatusNoContent)
		return
	}

	// Set application default header values.
	c.Header().Set("Content-Type", "application/json")

	// Load any user defined header values.
	if app.userHeaders != nil {
		for key, value := range app.userHeaders {
			log.User("startup", "Init", "Setting user headers : %s:%s", key, value)
			c.Header().Set(key, value)
		}
	}

	c.WriteHeader(code)

	// Marshal the data into a JSON string.
	jsonData, err := json.Marshal(data)
	if err != nil {
		jsonData = []byte("{}")
	}

	// Look for a JSONP marker
	if cb := c.Request.URL.Query().Get("callback"); cb != "" {

		// We need to wrap the result in a function call.
		// callback_value({"data_1": "hello world", "data_2": ["the","sun","is","shining"]});
		fmt.Fprintf(c, "%s(%s)", cb, string(jsonData))

		log.User(c.SessionID, "api : Respond", "Completed")
		return
	}

	// We can send the result straight through.
	fmt.Fprintf(c, string(jsonData))

	log.User(c.SessionID, "api : Respond", "Completed")
}

// Midware handles the request logging.
func Midware(next web.Handler) web.Handler {

	// Create the handler that will be attached in the middleware chain.
	h := func(c *web.Context) error {

		log.User(c.SessionID, "log : Midware", "Started : Method[%s] URL[%s] RADDR[%s]", c.Request.Method, c.Request.URL.Path, c.Request.RemoteAddr)

		if err := next(c); err != nil {
			return err
		}

		log.User(c.SessionID, "log : Midware", "Completed : Status[%d] Duration[%s]", c.Status, time.Since(c.Now))

		return nil
	}

	return h
}

// Respond sends JSON to the client.
// If code is StatusNoContent, v is expected to be nil.
func (c *Context) Respond(data interface{}, code int) {
	log.User(c.SessionID, "api : Respond", "Started : Code[%d]", code)

	c.Status = code

	if code == http.StatusNoContent {
		c.WriteHeader(http.StatusNoContent)
		return
	}

	// Set application default header values.
	c.Header().Set("Content-Type", "application/json")

	// Load any user defined header values.
	if app.userHeaders != nil {
		for key, value := range app.userHeaders {
			log.User("startup", "Init", "Setting user headers : %s:%s", key, value)
			c.Header().Set(key, value)
		}
	}

	c.WriteHeader(code)

	// Look for a JSONP marker
	if cb := c.Request.URL.Query().Get("callback"); cb != "" {

		// We need to wrap the result in a function call.
		// callback_value({"data_1": "hello world", "data_2": ["the","sun","is","shining"]});
		b := bytes.NewBufferString(cb + "(")
		json.NewEncoder(b).Encode(data)
		b.WriteString(")")
		fmt.Fprintf(c, b.String())

	} else {

		// We can send the result straight through.
		json.NewEncoder(c).Encode(data)

	}

	log.User(c.SessionID, "api : Respond", "Completed")
}

// ExampleDev shows how to use the log package.
func ExampleDev(t *testing.T) {
	// Init the log package for stdout. Hardcode the logging level
	// function to use USER level logging.
	log.Init(os.Stdout, func() int { return log.USER })

	// Write a simple log line with no formatting.
	log.User("context", "ExampleDev", "This is a simple line with no formatting")

	// Write a simple log line with formatting.
	log.User("context", "ExampleDev", "This is a simple line with no formatting %d", 10)

	// Write a message error for the user.
	log.Error("context", "ExampleDev", errors.New("A user error"), "testing error")

	// Write a message error for the user with formatting.
	log.Error("context", "ExampleDev", errors.New("A user error"), "testing error %s", "value")

	// Write a message error for the developer only.
	log.Dev("context", "ExampleDev", "Formatting %v", 42)
}

func importFromAPI(collections []string) {

	pageAfter := ""
	log.User(uuid, "sponge.importFromAPI", "### Reading data from API. \n")

	api, ok := dbsource.(source.API)
	if !ok {
		err := fmt.Errorf("Error asserting sourcer into source.API.")
		log.Error(uuid, "sponge.importFromAPI", err, "Asserting type for source.API")
	}

	var err error
	var data []map[string]interface{}
	var nextPageAfter string
	var pollingInterval time.Duration

	if pollEnvVar, err := strconv.Atoi(os.Getenv("POLLING_INTERVAL")); err != nil || pollEnvVar == 0 {
		pollingInterval = time.Duration(options.TimeWaiting) * time.Second
	} else {
		pollingInterval = time.Duration(pollEnvVar) * time.Second
	}

	for true {
		data, nextPageAfter, err = api.GetFireHoseData(pageAfter)
		if err != nil {
			log.Error(uuid, "sponge.importFromAPI", err, "Getting data from API")
			return
		}

		if data != nil {
			processAPI(collections, data)
			pageAfter = nextPageAfter
		}

		if data == nil {
			log.User(uuid, "sponge.importFromAPI", "Waiting %s seconds for more data.", pollingInterval)
			time.Sleep(pollingInterval) // sleep timeWaiting seconds
		}
	}

}

// CreateIndex will read the strategy file and create index that are mentioned there for each collection
func CreateIndex(collection string) {

	log.User(uuid, "sponge.createindex", "###  Create Index.")

	//create index for everybody
	if collection == "" {
		// get data from strategy file
		tables := fiddler.GetCollections()

		// for each table
		for t := range tables {
			log.User(uuid, "sponge.createindex", "### Create index for collection %s.", tables[t])
			coral.CreateIndex(tables[t])
		}
		return
	}

	log.User(uuid, "sponge.createindex", "### Create index for collection %s.", collection)
	//create index only for collection
	coral.CreateIndex(collection)
}

func main() {
	log.User("startup", "Init", "Revision     : %q", GitRevision)
	log.User("startup", "Init", "Version      : %q", GitVersion)
	log.User("startup", "Init", "Build Date   : %q", BuildDate)
	log.User("startup", "Init", "Int Version  : %q", IntVersion)
	log.User("startup", "Init", "Go Version   : %q", runtime.Version())
	log.User("startup", "Init", "Go Compiler  : %q", runtime.Compiler)
	log.User("startup", "Init", "Go ARCH      : %q", runtime.GOARCH)
	log.User("startup", "Init", "Go OS        : %q", runtime.GOOS)

	handlers.Version.GitRevision = GitRevision
	handlers.Version.GitVersion = GitVersion
	handlers.Version.BuildDate = BuildDate
	handlers.Version.IntVersion = IntVersion

	// These are the absolute read and write timeouts.
	const (

		// ReadTimeout covers the time from when the connection is accepted to when the
		// request body is fully read.
		readTimeout = 10 * time.Second

		// WriteTimeout normally covers the time from the end of the request header read
		// to the end of the response write.
		writeTimeout = 30 * time.Second
	)

	host := cfg.MustString(cfgHost)

	log.User("startup", "Init", "Binding web service to %s", host)

	if err := web.Run(host, routes.API(), readTimeout, writeTimeout); err != nil {
		log.Error("shutdown", "Init", err, "App Shutdown")
		os.Exit(1)
	}

	log.User("shutdown", "Init", "App Shutdown")
}