// Init ...
func (rthink *RethinkDBImpl) Init(conntype string, dbhost string, username string, password string, port int, dbname string, log log.Logger, hosts ...string) error {
	var retErr error
	if conntype == "cluster" {
		session, err := r.Connect(r.ConnectOpts{
			Addresses: hosts,
			Database:  dbname,
			AuthKey:   password,
		})
		retErr = err
		if err != nil {
			log.Crit("Got error when connect database, the error is '%v'", err)
			return retErr
		}
		rthink.Session = session
	} else {
		session, err := r.Connect(r.ConnectOpts{
			Address:  dbhost,
			Database: dbname,
			AuthKey:  password,
		})
		retErr = err
		if err != nil {
			log.Crit("Got error when connect database, the error is '%v'", err)
			return retErr
		}
		rthink.Session = session
	}
	fmt.Println("Rethink Connected")
	return nil
}

func init() {

	Log = logrus.New()

	session, err := r.Connect(r.ConnectOpts{
		Address:  "172.17.8.150:32768",
		Database: "gettingstarted",
	})
	if err != nil {
		log.WithFields(log.Fields{
			"Error": err,
		}).Error("Error getting rethink")
	}

	// if err = r.DbCreate("gettingstarted").Run(session).Exec(); err != nil {
	resp, err := r.DBCreate("gettingstarted").RunWrite(session)
	if err != nil {
		log.WithFields(log.Fields{
			"Error": err,
		}).Error("Error creating db")
	}
	fmt.Printf("RESP: %+v\n", resp.DBsCreated)

	// if err = r.TableCreate("bookmarks").Run(session).Exec(); err != nil {
	resp, err = r.DB("gettingstarted").TableCreate("bookmarks").RunWrite(session)
	if err != nil {
		log.WithFields(log.Fields{
			"Error": err,
		}).Error("Error creating table")
	}
	fmt.Printf("RESP %+v\n", resp.TablesCreated)

}

func New(conf Conf) (*Engine, error) {
	if len(conf.DBAddress) == 0 {
		return nil, ErrInvalidAddress
	}
	if len(conf.DBDatabase) == 0 {
		return nil, ErrInvalidDB
	}

	logger := log.New(os.Stdout, "[entrapped models]", log.Ldate|log.Ltime|log.Lshortfile)

	session, sessionErr := r.Connect(r.ConnectOpts{
		Address:  conf.DBAddress,
		Database: conf.DBDatabase,
		AuthKey:  conf.DBAuthKey,
		Timeout:  conf.DBTimeout,
		MaxIdle:  conf.DBMaxIdle,
		MaxOpen:  conf.DBMaxOpen,
	})

	if sessionErr != nil {
		logger.Println(sessionErr)
		return nil, ErrRethinkConn
	}

	return &Engine{logger, r.DB(conf.DBDatabase), session}, nil
}

func newLogger() (*Logger, error) {
	client := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})

	pubsub, err := client.Subscribe("messages")
	session, err := rdb.Connect(rdb.ConnectOpts{
		Address: "localhost:28015",
	})

	if err != nil {
		panic(err)
	}

	logger := &Logger{
		Redis:     client,
		Pubsub:    pubsub,
		DbSession: session,
		messages:  make(chan string),
	}

	go logger.handleMessages()

	return logger, err
}

// InitDBs prepares a RethinkDB instance to be used by rethinkClient.
// rethinkClients will error if they are pointed at databases that haven't had
// InitDBs run on them
// InitDBs should only be run once per instance of RethinkDB, it will error if
// it's called a second time.
func InitDBs(address string, databaseName string) error {
	session, err := gorethink.Connect(gorethink.ConnectOpts{Address: address})
	if err != nil {
		return err
	}
	if _, err := gorethink.DBCreate(databaseName).RunWrite(session); err != nil {
		return err
	}
	for _, table := range tables {
		tableCreateOpts, ok := tableToTableCreateOpts[table]
		if ok {
			if _, err := gorethink.DB(databaseName).TableCreate(table, tableCreateOpts...).RunWrite(session); err != nil {
				return err
			}
		} else {
			if _, err := gorethink.DB(databaseName).TableCreate(table).RunWrite(session); err != nil {
				return err
			}
		}
	}
	for table, indexes := range tableToIndexes {
		for _, index := range indexes {
			if _, err := gorethink.DB(databaseName).Table(table).IndexCreate(index).RunWrite(session); err != nil {
				return err
			}
		}
	}
	return nil
}

// Connect to the database.
func Connect() (err error) {
	// Connext to the database server.
	Session, err = r.Connect(r.ConnectOpts{
		Address:   settings.Settings.DBAddress,
		Addresses: settings.Settings.DBAddresses,
		Database:  settings.Settings.DBName,
		MaxIdle:   settings.Settings.DBMaxIdle,
		MaxOpen:   settings.Settings.DBMaxOpen,
		Timeout:   settings.Settings.DBTimeout,
		AuthKey:   settings.Settings.DBAuthKey,

		// When DiscoverHosts is true any nodes are added to the cluster after
		// the initial connection then the new node will be added to the pool of
		// available nodes used by GoRethink. Unfortunately the canonical address
		// of each server in the cluster MUST be set as otherwise clients will
		// try to connect to the database nodes locally. For more information
		// about how to set a RethinkDB servers canonical address set
		// this page http://www.rethinkdb.com/docs/config-file/.
		DiscoverHosts: true,
	})
	if err != nil {
		return fmt.Errorf("failed to connect to database: %v", err)
	}

	return nil
}

func main() {

	dbAddress := util.GetKeyValue("RethinkDB", "address")

	var err error
	session, err = db.Connect(db.ConnectOpts{
		Address:  dbAddress,
		Database: "travelPlanning",
	})
	if err != nil {
		log.Fatalln(err.Error())
	}
	log.Println("Connected to db on: " + dbAddress)

	tasks := mq.NewMQ()
	tasks.NewQueue("TripWorkerQueue", "Trip")
	msgs := tasks.GetMessages("TripWorkerQueue")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			var message mq.Message
			json.Unmarshal(d.Body, &message)
			processMessage(&message)
			tasks.PublishMessage(message, d.ReplyTo, d.CorrelationId, "")
		}
	}()

	log.Printf(" [*] Awaiting for requests")
	<-forever

}

func (e *Exporter) scrape(scrapes chan<- scrapeResult) {

	defer close(scrapes)

	now := time.Now().UnixNano()
	e.totalScrapes.Inc()

	sess, err := r.Connect(r.ConnectOpts{
		Addresses: e.addrs,
		Database:  "rethinkdb",
		AuthKey:   e.auth,
	})

	errCount := 0
	if err == nil {
		if err := extractAllMetrics(sess, scrapes); err != nil {
			errCount++
		}
		sess.Close()
	}

	if err != nil {
		log.Printf("scrape err: %s", err)
	}
	e.scrapeError.Set(float64(errCount))
	e.duration.Set(float64(time.Now().UnixNano()-now) / 1000000000)
}

func NewRethinkDBAdapter(c *CAS) (*RethinkDBAdapter, error) {
	// Database setup
	dbSession, err := r.Connect(r.ConnectOpts{
		Address:  c.Config["dbHost"],
		Database: c.Config["dbName"],
	})
	if err != nil {
		return nil, err
	}

	// Create the adapter
	adapter := &RethinkDBAdapter{
		session:              dbSession,
		dbName:               c.Config["dbName"],
		ticketsTableName:     "tickets",
		ticketsTableOptions:  nil,
		servicesTableName:    "services",
		servicesTableOptions: &r.TableCreateOpts{PrimaryKey: "name"},
		usersTableName:       "users",
		usersTableOptions:    &r.TableCreateOpts{PrimaryKey: "email"},
		apiKeysTableName:     "api_keys",
		apiKeysTableOptions:  &r.TableCreateOpts{PrimaryKey: "key"},
		LogLevel:             c.Config["logLevel"],
	}

	return adapter, nil
}

//NewDBSession creates a new session to manage database data
func NewDBSession(database string) *r.Session {
	conn, err := r.Connect(r.ConnectOpts{
		Address:  GetKeyValue("database", "address"),
		Database: database,
		MaxOpen:  40,
	})
	if err != nil {
		log.Fatalln(err)
		return nil
	}
	log.Infof("Connected to RethinkDB %s", GetKeyValue("database", "address"))

	status, err := createDB(database, conn)
	if err != nil {
		log.Fatalln(err)
	} else {
		log.Info(status)
	}

	tables := []string{"Trip", "Place", "User"}

	total, err := createTables(database, tables, conn)
	if err != nil {
		log.Fatalln(err)
	} else {
		log.Infof("%d new table(s) created", total)
	}

	return conn
}

func init() {
	flag.Parse()
	err := gcfg.ReadFileInto(&cfg, *configFile)
	if err != nil {
		log.Fatal(err)
	}

	session, err = r.Connect(r.ConnectOpts{
		Address:  cfg.Database.Host + ":" + cfg.Database.Port, //localhost:28015
		Database: cfg.Database.DB,                             //DB: cats
	})
	if err != nil {
		log.Fatal("Could not connect")
	}
	res, err := r.DBCreate(cfg.Database.DB).RunWrite(session)
	if err != nil {
		log.Println(err.Error())
	}

	fmt.Printf("%d DB created\n", res.DBsCreated)
	r.DB(cfg.Database.DB).TableCreate("instacat").Run(session)
	log.Println("Create table instacat.")
	r.Table("instacat").IndexCreate("time").Run(session)
	log.Println("Create index time.")
	r.Table("instacat").IndexCreate("place", r.IndexCreateOpts{Geo: true}).Run(session)
	log.Println("Create index place.")
}

func initializeDB() {
	var err error
	session, err = gorethink.Connect(gorethink.ConnectOpts{Address: config.DBHost, Database: config.DBName})
	if err != nil {
		logrus.Fatal("db:", err)
	}
}

func (c *DBConfig) Setup() *DBSession {
	if addr := os.Getenv(dbAddressEnv); addr != "" {
		c.Address = addr
	}
	if name := os.Getenv(dbNameEnv); name != "" {
		c.Name = name
	}
	if table := os.Getenv(gameTableEnv); table != "" {
		c.GameTable = table
	}

	sess, err := r.Connect(r.ConnectOpts{
		Address:  c.Address,
		Database: c.Name,
	})
	if err != nil {
		log.Fatalf("RethinkDB Connection error: %s\n", err.Error())
	}

	// set session in DBConfig
	dbs := &DBSession{S: sess, Config: *c}

	// ensure named gameTable exists in database ...
	dbs.CreateTable(c.GameTable)

	// ... with the correct indexes
	dbs.CreateIndexes(c.GameTable, c.Indexes)

	return dbs
}

func main() {
	log.SetFlags(log.LstdFlags | log.Lshortfile)

	flag.Parse()

	// Connect to RethinkDB
	var err error
	session, err = r.Connect(r.ConnectOpts{
		Address: *rethinkAddress,
	})
	if err != nil {
		log.Fatal(err)
	}

	// Create the database and tables
	r.DbCreate(*rethinkName).Exec(session)
	r.Db(*rethinkName).TableCreate("invites").Exec(session)
	r.Db(*rethinkName).Table("invites").IndexCreate("email").Exec(session)
	r.Db(*rethinkName).Table("invites").IndexCreate("name").Exec(session)

	// Add a CORS middleware
	goji.Use(cors.New(cors.Options{
		AllowCredentials: true,
	}).Handler)

	// Add routes to goji
	goji.Get("/", index)
	goji.Post("/check", check)
	goji.Post("/free", free)
	goji.Post("/create", create)

	// Start the server
	goji.Serve()
}

func initdb() {
	// retries
	for i := 0; i < 5; i++ {
		s, err := rethink.Connect(rethink.ConnectOpts{
			Address:  dbAddr,
			Database: dbName,
		})
		if err != nil {
			log.Printf("unable to connect; retrying: %s", err)
			time.Sleep(2 * time.Second)
			continue
		}

		session = s
	}

	if session == nil {
		log.Fatalf("unable to get database connection")
	}

	// Create database
	rethink.DBCreate(dbName).Run(session)

	// Check if table exists
	_, err := rethink.Table(tblVisits).Run(session)
	if err != nil {
		// If not, create it
		if _, err = rethink.DB(dbName).TableCreate(tblVisits).Run(session); err != nil {
			log.Fatalf("error creating table: %s", err)
		}
	}
}

func add(config Config) {
	conn, err := r.Connect(r.ConnectOpts{
		Address:  config.RethinkDBAddress,
		Database: config.RethinkDBDatabase,
		AuthKey:  config.RethinkDBAuthkey,
		TLSConfig: &tls.Config{
			InsecureSkipVerify: true,
		},
	})
	if err != nil {
		log.Fatal(conn, err)
	}

	type URLentry struct {
		ID  string `gorethink:"id,omitempty"`
		URL string `gorethink:"url"`
	}

	r.DB(config.RethinkDBDatabase).Table("urls").Insert(URLentry{URL: *urlarg}).RunWrite(conn)
	if err != nil {
		log.Fatal(err)
	}

	conn.Close()
}

func main() {
	r := gin.Default()
	r.Use(RequestIdMiddleware())
	DB, err := db.Connect(db.ConnectOpts{
		Address:  fmt.Sprintf("%s:%d", os.Getenv("RETHINKDB_PORT_28015_TCP_ADDR"), 28015),
		Database: "test",
	})
	if err != nil {
		log.Fatalln(err.Error())
	}
	db.TableCreate("users").RunWrite(DB)
	db.TableCreate("resources").RunWrite(DB)
	store, err := sessions.NewRedisStore(10, "tcp", fmt.Sprintf("%s:%d", os.Getenv("REDIS_1_PORT_6379_TCP_ADDR"), 6379), "", []byte(os.Getenv("REDIS_SECRET")))
	if err != nil {
		log.Fatalln(err.Error())
	}
	gob.Register(models.User{})
	r.Use(sessions.Sessions("session", store))
	api.Init(r, DB, RequestAuthMiddleware)
	login.Init(r, DB)
	r.Any("/ping", func(c *gin.Context) {
		if DB.IsConnected() {
			c.String(200, "ok")
		} else {
			c.String(500, "not ok")
		}
	})
	r.Run(":8080") // listen and serve on 0.0.0.0:8080
}

func InitDB() *rethink.Session {

	session, err := rethink.Connect(map[string]interface{}{
		"address":     "localhost:8080", //os.Getenv("RETHINKDB_URL"),
		"database":    "test",
		"maxIdle":     10,
		"idleTimeout": time.Second * 10,
	})

	if err != nil {
		log.Println(err)
	}

	err = rethink.DbCreate("test").Exec(session)
	if err != nil {
		log.Println(err)
	}

	_, err = rethink.Db("test").TableCreate("articles").RunWrite(session)
	if err != nil {
		log.Println(err)
	}

	return session
}

func init() {
	var err error
	session, err = gorethink.Connect(gorethink.ConnectOpts{
		Address: "127.0.0.1:28015",
	})
	if err != nil {
		panic(err)
	}

	key2find = uniuri.New()

	// Create a new table
	gorethink.Db("test").TableDrop("benchmark_keys_list").Run(session)
	gorethink.Db("test").TableCreate("benchmark_keys_list").Run(session)

	var klist []*KeysList

	// Populate with sample data
	for n := 0; n < 300; n++ {
		keys := rndStringSlice(999)
		keys = randomlyInsert(keys, key2find)

		y := uniuri.New()
		if n == 153 {
			table2search = y
		}

		klist = append(klist, &KeysList{
			ID:    y,
			Voted: keys,
		})
	}

	gorethink.Db("test").Table("benchmark_keys_list").Insert(klist).Run(session)
}

func main() {
	flag.Parse()

	var err error
	session, err := r.Connect(r.ConnectOpts{
		Hosts:    strings.Split(*rethinkdbName, ","),
		Database: *rethinkdbDatabase,
	})
	if err != nil {
		log.Fatal(err)
	}

	goji.Get("/accounts", listAccounts)
	goji.Post("/accounts", createAccount)
	goji.Get("/accounts/:id", getAccount)
	goji.Put("/accounts/:id", updateAccount)
	goji.Delete("/accounts/:id", deleteAccount)

	goji.Get("/accounts/:id/tokens", listAccountTokens)
	goji.Delete("/accounts/:id/tokens", deleteAccountToken)

	goji.Get("/tokens", listTokens)
	goji.Post("/tokens", createToken)
	goji.Get("/tokens/:id", getToken)
	goji.Put("/tokens/:id", updateToken)
	goji.Delete("/tokens/:id", deleteToken)

	goji.Serve()
}