// This handles the timing of re-checking configuration through
// a Go channel that is never closed, hence endlessly repeating
func configTimer(interval *string) chan struct{} {

	go loadBrokers()

	dur, err := time.ParseDuration(*interval)
	if err != nil {
		dur = 1 * time.Minute
	}

	log.Info(fmt.Sprintf("Re-loading configuration every %s", dur))

	ticker := time.NewTicker(dur)
	quit := make(chan struct{})
	go func() {
		for {
			select {
			case <-ticker.C:
				log.Info("Re-checking/loading brokers config")
				loadBrokers()
			case <-quit:
				ticker.Stop()
				return
			}
		}
	}()

	return quit

}

func TestAudioToBracelet(t *testing.T) {
	llog.Info("Start test")
	audio, err := s.GetAudio("./test.amr")
	if err != nil {
		llog.Error("Erorr", "err", err)
		t.FailNow()
	}

	length := len(audio)
	// Divide file in parts
	var parts int
	if length%s.BufferSize == 0 {
		parts = length / s.BufferSize
	} else {
		parts = length/s.BufferSize + 1
	}

	from := Expect{"ARA": Data{"", 1, "666"}, "ASA": Data{"", 1, "666"}, "ARD": Data{"", 1, ""}}
	to := Expect{"RAU": Data{"", 1, ""}, "SAE": Data{"", 1, ""}} // "SAU": Data{"", parts, ""},

	br := Registration(from, to, t)
	br.Lock()
	br.Commands["666"] = audio
	br.Unlock()

	command := fmt.Sprintf("RAU&M%v&%v&%v&%v&%v&%v", br.Clients[0], 666, length, parts, 10, "DF")
	br, from, to = SendInstruction(br, from, to, command, "ARD", false, t)

	if !(from.Done() && to.Done()) {
		t.Fail()
	}
	br.Logger.Info("End of the test")
	server.RemoveBracelet(TestingID)
	llog.Info("End test")
}

func TestEmoji(t *testing.T) {
	llog.Info("Start test")
	emoji := []byte("47494638396160004000b30f00ee124bf0124cdc1046090103e611492a030de111471c0209eb114a110105f3124df2124cb2ed87f4124d000000ffffff21ff0b4e45545343415045322e30030100000021ff0b584d502044617461584d503c3f787061636b657420626567696e3d22efbbbf222069643d2257354d304d7043656869487a7265537a4e54637a6b633964223f3e203c783a786d706d65746120786d6c6e733a783d2261646f62653a6e733a6d6574612f2220783a786d70746b3d2241646f626520584d5020436f726520352e332d633031312036362e3134353636312c20323031322f30322f30362d31343a35363a32372020202020202020223e203c7264663a52444620786d6c6e733a7264663d22687474703a2f2f7777772e77332e6f72672f313939392f30322f32322d7264662d73796e7461782d6e7323223e203c7264663a4465736372697074696f6e207264663a61626f75743d222220786d6c6e733a786d704d4d3d22687474703a2f2f6e732e61646f62652e636f6d2f7861702f312e302f6d6d2f2220786d6c6e733a73745265663d22687474703a2f2f6e732e61646f62652e636f6d2f7861702f312e302f73547970652f5265736f75726365526566232220786d6c6e733a786d703d22687474703a2f2f6e732e61646f62652e636f6d2f7861702f312e302f2220786d704d4d3a4f726967696e616c446f63756d656e7449443d22786d702e6469643a39413539433632423234343245353131423537354336374132304541303631392220786d704d4d3a446f63756d656e7449443d22786d702e6469643a39343041303044413432363531314535423537434235333434323437433846392220786d704d4d3a496e7374616e636549443d22786d702e6969643a39343041303044393432363531314535423537434235333434323437433846392220786d703a43726561746f72546f6f6c3d2241646f62652050686f746f73686f7020435336202857696e646f777329223e203c786d704d4d3a4465726976656446726f6d2073745265663a696e7374616e636549443d22786d702e6969643a3839443236434345353034324535313139303338444438383144413042333837222073745265663a646f63756d656e7449443d22786d702e6469643a3938353943363242323434324535313142353735433637413230454130363139222f3e203c2f7264663a4465736372697074696f6e3e203c2f7264663a5244463e203c2f783a786d706d6574613e203c3f787061636b657420656e643d2272223f3e01fffefdfcfbfaf9f8f7f6f5f4f3f2f1f0efeeedecebeae9e8e7e6e5e4e3e2e1e0dfdedddcdbdad9d8d7d6d5d4d3d2d1d0cfcecdcccbcac9c8c7c6c5c4c3c2c1c0bfbebdbcbbbab9b8b7b6b5b4b3b2b1b0afaeadacabaaa9a8a7a6a5a4a3a2a1a09f9e9d9c9b9a999897969594939291908f8e8d8c8b8a898887868584838281807f7e7d7c7b7a797877767574737271706f6e6d6c6b6a696867666564636261605f5e5d5c5b5a595857565554535251504f4e4d4c4b4a494847464544434241403f3e3d3c3b3a393837363534333231302f2e2d2c2b2a292827262524232221201f1e1d1c1b1a191817161514131211100f0e0d0c0b0a090807060504030201000021f9040528000f002c00000000600040000004ffd0c949abbd38ebcdbbff60288e64699e68aaae2cdabc4000bccdd0de6881d03c5fe0c0d2acd73b048f15226fb250d290d046d3399b381550e4ceb9941094d9e396bb4c1888615cc231562c020b8553327ea55b2fa366daeddd592f3f127e147c")

	length := len(emoji)
	// Divide file in parts
	var parts int
	if length%s.BufferSize == 0 {
		parts = length / s.BufferSize
	} else {
		parts = length/s.BufferSize + 1
	}

	from := Expect{"ARA": Data{"", 1, "666"}, "ASA": Data{"", 1, "666"}}
	to := Expect{"RAU": Data{"", 1, ""}, "SAE": Data{"", 1, ""}}

	br := Registration(from, to, t)
	br.Lock()
	br.Commands["666"] = emoji
	br.Unlock()

	command := fmt.Sprintf("RAU&M%v&%v&%v&%v&%v&%v", br.Clients[0], 666, length, parts, 1, "EI")
	br, from, to = SendInstruction(br, from, to, command, "ASA", false, t)

	if !(from.Done() && to.Done()) {
		t.Fail()
	}
	br.Logger.Info("End of the test")
	server.RemoveBracelet(TestingID)
	llog.Info("End test")
}

// Initializes database configuration.
// Sets up the psuedo-orm's table mapping and creates tables if not already existing.
func initDb() {

	if dbmap != nil {
		log.Info("Reusing existing database connection")
		return
	}

	var err error

	log.Info("Initializing database connection")

	db, err = sql.Open("mysql", dsn)
	db.SetMaxOpenConns(dbMaxConn)

	fatalErr(err, "Could not connect to database")

	dbmap = &gorp.DbMap{Db: db, Dialect: gorp.MySQLDialect{"InnoDB", "UTF8"}}

	dbmap.AddTableWithName(Broker{}, "mqtt_logger_config").SetKeys(true, "Id")
	dbmap.AddTableWithName(Topic{}, "mqtt_logger_topics").SetKeys(true, "Id")
	dbmap.AddTableWithName(Message{}, "mqtt_logger_messages").SetKeys(true, "Id")
	err = dbmap.CreateTablesIfNotExists()
	fatalErr(err, "Could not create database tables")

	log.Info(fmt.Sprintf("Connected successfully to %s", dsn))

}

func (s *CabifyService) Run(cfg config.Config) error {

	cabifyResource := resources.CabifyResource{}

	log.Info("Starting Cabify Store service")

	cabifyResource.Cfg = cfg

	r := gin.Default()

	// Admin config route
	routes := r.Group("/store")
	{
		routes.POST("/totalprice", (&cabifyResource).OrderSum)
	}

	//GIN release mode, either debug, test or release (DebugMode, TestMode, ReleaseMode)
	//by default, runs in releasemode
	gin.SetMode(gin.DebugMode)
	log.Info("Cabify Store service run in", "mode", gin.Mode())

	r.Run(cfg.SvcHost)

	return nil
}

// Handles the subscribing and unsubscribing of topics for
// a given broker. Is called on connect and when re-loading
// configuration from the database.
func (b Broker) updateSubscriptions(c *mqtt.Client) {

	topics, err := b.GetTopics()
	checkErr(err, "Could not get topics")

	// Check for new entries to subscribe.
	for t, _ := range topics {

		if !isValueInList(t, status[b.Server].Topics) {

			log.Info(fmt.Sprintf("Subscribing to %s on %s", t, b.Server))

			// If don't do it this way, get a cannot assign to error.
			s := status[b.Server]
			s.Topics = append(status[b.Server].Topics, t)
			status[b.Server] = s

			go func() {
				token := c.Subscribe(t, 1, b.OnMessage)
				if token.Wait() && token.Error() != nil {
					delete(status, b.Server)
					checkErr(token.Error(), "Error subscribing to topic(s)")
				}
			}()

		}

	}

	// Now check to unsubscribe if topic has been deleted.
	// Create a new empty slice for topics, and if found, we
	// append to it, if not found just skip it.
	newTopics := []string{}
	for _, t := range status[b.Server].Topics {

		found := false
		for k, _ := range topics {
			if t == k {
				found = true
			}
		}

		// No longer subscribed, so unsubcribe now
		if !found {
			log.Info(fmt.Sprintf("Unsubscribing from %s on %s", t, b.Server))
			token := status[b.Server].Client.Unsubscribe(t)
			checkErr(token.Error(), "Error unsubscribing")
		} else {
			newTopics = append(newTopics, t)
		}

	}

	// Now update the status variable with the new Topics list for this server.
	s := status[b.Server]
	s.Topics = newTopics
	status[b.Server] = s

}

func main() {
	options := cfg.GetOptions()

	log.Info("Server is ready!", "port", options.Port, "token", options.Token)
	log.Info(fmt.Sprintf("Listening for post requests on http://localhost:%s/events", options.Port))
	log.Info(fmt.Sprintf("SSE streaming avaliable on http://localhost:%s/stream", options.Port))

	es := eventsource.New(
		eventsource.DefaultSettings(),
		func(req *http.Request) [][]byte {
			return [][]byte{
				[]byte("X-Accel-Buffering: no"),
				[]byte("Access-Control-Allow-Origin: *"),
				[]byte("Cache-Control: no-cache"),
			}
		},
	)

	defer es.Close()

	r := mux.NewRouter()

	r.Handle("/stream", es)

	r.HandleFunc("/event/{event}/{id}", func(w http.ResponseWriter, r *http.Request) {
		if r.Method == "POST" {
			if token, ok := r.Header["X-Token"]; ok {
				if token[0] == options.Token {
					event := mux.Vars(r)["event"]
					id := mux.Vars(r)["id"]

					data, err := ioutil.ReadAll(r.Body)
					if err != nil {
						log.Error("Cannot read body", "err", err)
					}

					es.SendEventMessage(string(data), event, id)
					log.Info("Message has been sent", "id", id, "event", event)

				} else {
					log.Warn("The request has wrong token", "token", token[0])
					http.Error(w, "The request has wrong token", http.StatusUnauthorized)
				}
			} else {
				log.Warn("The request doesn't contain authentication token")
				http.Error(w, "The request doesn't contain authentication token", http.StatusUnauthorized)
			}
		} else {
			log.Warn("Received wrong http request")
			http.Error(w, "POST requests only", http.StatusMethodNotAllowed)
		}
	})

	r.Handle("/", http.FileServer(http.Dir("./public")))
	http.ListenAndServe(fmt.Sprintf(":%s", options.Port), r)
}

// The main function. All Go programs start here.
func main() {

	// This is just to keep things open. Could be used later on to output
	// stored data, reload configuration, etc.
	http.HandleFunc("/", addDefaultHeaders(indexHandler))
	http.HandleFunc("/status", addDefaultHeaders(statusHandler))

	log.Info(fmt.Sprintf("About to listen on %s", listenStr))
	log.Info(fmt.Sprintf("LoggerID: %v", loggerID))

	err := http.ListenAndServe(listenStr, nil)
	fatalErr(err, "Server failed to start")

}

func (tr *CabifyResource) OrderSum(c *gin.Context) {
	var data api.Order
	var arr []string
	var summ float32

	summ = 0
	log.Info("Started")

	err := c.Bind(&data)
	if err != nil {
		c.JSON(400, api.NewError("problem decoding body"))
		return
	}

	log.Info("Items parcing, data is: " + data.Items)

	arr = strings.Split(data.Items, ",")
	if arr == nil {
		log.Error("Empty order list, please fill it")
	}

	var VOUCHER_cnt = strings.Count(data.Items, api.VOUCHER)
	var TSHIRT_cnt = strings.Count(data.Items, api.TSHIRT)
	var MUG_cnt = strings.Count(data.Items, api.MUG)

	log.Info("The number of " + api.VOUCHER + " occurence = " + strconv.Itoa(int(VOUCHER_cnt)))
	log.Info("The number of " + api.TSHIRT + " occurence = " + strconv.Itoa(int(TSHIRT_cnt)))
	log.Info("The number of " + api.MUG + " occurence = " + strconv.Itoa(int(MUG_cnt)))

	if TSHIRT_cnt >= 3 {
		summ = summ + float32(TSHIRT_cnt*api.CFO_DISCONT_TSHIRT_PRICE)
	} else if (TSHIRT_cnt < 3) && (TSHIRT_cnt > 0) {
		summ = summ + api.TSHIRT_PRICE
	}

	if VOUCHER_cnt > 0 {
		summ = summ + api.VOUCHER_PRICE

		if math.Mod(float64(MUG_cnt), 2) == 0 {
			summ = summ + float32(int(VOUCHER_cnt/2))*api.VOUCHER_PRICE
		}
	}

	summ = summ + float32(MUG_cnt)*api.MUG_PRICE

	c.JSON(200, gin.H{
		"Order Summ": fmt.Sprintf("%.2f", summ),
		"Status":     "Success",
	})
}

func (e Expect) Check(msg, status string) {
	key := GetHeader(msg)
	if v, ok := e[key]; ok && v.Number != 0 {
		if v.CommID != "" && v.CommID != GetCommID(msg) {
			return
		}
		e[key] = Data{msg, v.Number - 1, v.CommID}
		if key != "SAU" {
			llog.Info(status, "Message", msg, "Left", e[key].Number)
		} else {
			llog.Info(status, "Message", strings.Join(strings.Split(msg, "&")[:3], "&"), "Left", e[key].Number)
		}
	}
}

func TestPush(t *testing.T) {
	llog.Info("Start test")
	from := Expect{"GET_FA_NAME_IMG": Data{"", 1, ""}}
	to := Expect{"PUS": Data{"", 1, ""}, "R_GET_FA_NAME_IMG": Data{"", 1, ""}}
	br := Registration(from, to, t)

	command := br.Push()
	br, from, to = SendInstruction(br, from, to, command, "GET_FA_NAME_IMG", false, t)

	if !(from.Done() && to.Done()) {
		t.Fail()
	}
	br.Logger.Info("End of the test")
	server.RemoveBracelet(TestingID)
	llog.Info("End test")
}

func TestCallPhone(t *testing.T) {
	llog.Info("Start test")
	from := Expect{"ACM": Data{"", 1, "666"}}
	to := Expect{"CMD": Data{"", 1, ""}}
	br := Registration(from, to, t)

	command := fmt.Sprintf(`CMD&%v&%v&kt*dhjt*%v*1*%v*`, from["ACM"].CommID, br.Clients[0], "89160252646", br.ID)
	br, from, to = SendInstruction(br, from, to, command, "ACM", true, t)

	if !(from.Done() && to.Done()) {
		t.Fail()
	}
	br.Logger.Info("End of the test")
	server.RemoveBracelet(TestingID)
	llog.Info("End test")
}

func TestMelody(t *testing.T) {
	llog.Info("Start test")
	from := Expect{"ARI": Data{"", 4, "666"}}
	to := Expect{"SRI": Data{"", 4, ""}}
	br := Registration(from, to, t)

	command := fmt.Sprintf(`SRI&%v&%v&3`, from["ARI"].CommID, br.Clients[0])
	br, from, to = SendInstruction(br, from, to, command, "ARI", true, t)

	from["ARI"] = Data{"", from["ARI"].Number, "667"}
	command = fmt.Sprintf(`SRI&%v&%v&1`, from["ARI"].CommID, br.Clients[0])
	br, from, to = SendInstruction(br, from, to, command, "ARI", true, t)

	from["ARI"] = Data{"", from["ARI"].Number, "668"}
	command = fmt.Sprintf(`SRI&%v&%v&0`, from["ARI"].CommID, br.Clients[0])
	br, from, to = SendInstruction(br, from, to, command, "ARI", true, t)

	from["ARI"] = Data{"", from["ARI"].Number, "669"}
	command = fmt.Sprintf(`SRI&%v&%v&2`, from["ARI"].CommID, br.Clients[0])
	br, from, to = SendInstruction(br, from, to, command, "ARI", true, t)

	if !(from.Done() && to.Done()) {
		t.Fail()
	}
	br.Logger.Info("End of the test")
	server.RemoveBracelet(TestingID)
}

func main() {
	// read environment vars
	common.ReadEnvironment()

	// initialise mux router
	router := mux.NewRouter()

	// setup statics
	poddy.CreateStaticBox(router)

	// setup watchdirectory
	if common.Watch != "" {
		check, err := os.Stat(common.Watch)
		if err != nil || !check.IsDir() {
			log.Error("error setting up watchdirectory (point to directory?)")
		} else {
			poddy.EnableWatchdirectory(common.Watch)
		}
	}

	// http handles
	router.HandleFunc("/", poddy.IndexPage)
	router.HandleFunc("/add-podcast", poddy.AddPodcast)
	router.HandleFunc("/feed", poddy.Feed)
	router.PathPrefix("/download").Handler(http.StripPrefix("/download", http.FileServer(http.Dir("storage/"))))

	// start server
	log.Info("poddy is running/listening", "host", common.Host, "port", common.Port)
	err := http.ListenAndServe(fmt.Sprintf("%s:%d", common.Host, common.Port), router)
	if err != nil {
		log.Crit("daemon could not bind on interface", "host", common.Host, "port", common.Port)
		os.Exit(1)
	}
}

func main() {
	var err error
	connPoolConfig := pgx.ConnPoolConfig{
		ConnConfig: pgx.ConnConfig{
			Host:     "127.0.0.1",
			User:     "jack",
			Password: "jack",
			Database: "url_shortener",
			Logger:   log.New("module", "pgx"),
		},
		MaxConnections: 5,
		AfterConnect:   afterConnect,
	}
	pool, err = pgx.NewConnPool(connPoolConfig)
	if err != nil {
		log.Crit("Unable to create connection pool", "error", err)
		os.Exit(1)
	}

	http.HandleFunc("/", urlHandler)

	log.Info("Starting URL shortener on localhost:8080")
	err = http.ListenAndServe("localhost:8080", nil)
	if err != nil {
		log.Crit("Unable to start web server", "error", err)
		os.Exit(1)
	}
}

func moveMP3toStorageDirectory(watchdir, storage string) {
	files, err := ioutil.ReadDir(watchdir)
	if err != nil {
		log.Error("scan error", "error", err)
		return
	}

	for _, file := range files {
		if !file.IsDir() {
			if strings.HasSuffix(file.Name(), "mp3") {
				source := filepath.Join(watchdir, file.Name())
				target := filepath.Join(storage, file.Name())

				// first copy file
				log.Info("about to copy file", "source", source, "target", target)
				err := copyFileContents(source, target)
				if err != nil {
					log.Error("copy error", "error", err)
					return
				}

				// then delete original file
				err = os.Remove(source)
				if err != nil {
					log.Error("error deleting source file", "error", err)
				}
			}
			log.Debug("only process mp3 files")
		}
	}
}

func main() {

	app := cli.NewApp()
	app.Name = "Cabify Store"
	app.Usage = "Cabify"
	app.Version = "0.0.1"

	app.Flags = []cli.Flag{
		cli.StringFlag{"config, c", "config.yaml", "config file to use", "APP_CONFIG"},
	}

	app.Commands = []cli.Command{
		{
			Name:  "server",
			Usage: "Run Cabify Store http server",
			Action: func(c *cli.Context) {
				cfg, err := getConfig(c)
				if err != nil {
					log.Error(err.Error())
					return
				}

				svc := service.CabifyService{}

				if err = svc.Run(cfg); err != nil {
					log.Error(err.Error())
				}

				log.Info("Cabify Store server started")
			},
		},
	}
	app.Run(os.Args)
}

func (a *albumJob) startDownloader() {
	defer a.downloadSync.Done()

	for item := range a.downloadQueue {
		buf := a.bufferPool.Get()

		log.Debug("Processing album item",
			"artist", item.Artist,
			"title", item.Title,
			"url", item.ItemURL,
		)

		err := downloadURL(buf, item.DownloadURL)
		if err != nil {
			log.Error("Unable to download item",
				"download_url", item.DownloadURL,
				"error", err,
			)
			return
		}
		a.zipQueue <- zipFile{
			Filename: item.Name(),
			Buffer:   buf,
		}
		log.Info("Processed album item",
			"download_url", item.DownloadURL,
			"filename", item.Name(),
		)
	}
}

// Callback when a message is received from a subscription.
// Right now, it inserts the message in the messages table of
// the database.
func (b Broker) OnMessage(client *mqtt.Client, msg mqtt.Message) {
	log.Info(fmt.Sprintf("Message received: %s", msg))

	err := b.Dbm.Insert(&Message{Topic: msg.Topic(), Data: string(msg.Payload()), Broker: b.Id, LoggerID: loggerID})
	checkErr(err, "Failed to insert message into database")

}

func main() {
	db, err := mgo.Dial(os.Getenv("MONGO_URL"))
	if err != nil {
		log.Error("Failed to connect to MongoDB", "error", err)
		os.Exit(1)
		return
	}
	defer db.Close()

	ops, errs := gtm.Tail(db, &gtm.Options{
		OpLogCollectionName: &OplogCollection,
		Filter: func(op *gtm.Op) bool {
			return op.Namespace == "dg.lists"
		},
	})

	for {
		select {
		case err = <-errs:
			log.Error("Error from oplog tail", "error", err)
		case <-ops:
			log.Info("Change event")
			Reload()
		}
	}
}