// CheckDBs checks that we have all the tables/indices we need
func CheckDBs(address string, databaseName string) error {
	session, err := connect(address)
	if err != nil {
		return err
	}

	for _, table := range tables {
		if _, err := gorethink.DB(databaseName).Table(table).Wait().RunWrite(session); err != nil {
			return err
		}
	}

	if _, err := gorethink.DB(databaseName).Table(jobInfosTable).IndexWait(pipelineNameIndex).RunWrite(session); err != nil {
		return err
	}

	if _, err := gorethink.DB(databaseName).Table(jobInfosTable).IndexWait(commitIndex).RunWrite(session); err != nil {
		return err
	}

	if _, err := gorethink.DB(databaseName).Table(jobInfosTable).IndexWait(pipelineNameAndCommitIndex).RunWrite(session); err != nil {
		return err
	}

	if _, err := gorethink.DB(databaseName).Table(pipelineInfosTable).IndexWait(pipelineShardIndex).RunWrite(session); err != nil {
		return err
	}

	return nil
}

// UserCreate creates a new user. Returns error and true if error is due to duplicate user name,
// false for any other error
func (a *RethinkDbAdapter) UserCreate(user *t.User) (error, bool) {
	// Save user's tags to a separate table to ensure uniquness
	// TODO(gene): add support for non-unique tags
	if user.Tags != nil {
		type tag struct {
			Id     string
			Source string
		}
		tags := make([]tag, 0, len(user.Tags))
		for _, t := range user.Tags {
			tags = append(tags, tag{Id: t, Source: user.Id})
		}
		res, err := rdb.DB(a.dbName).Table("tagunique").Insert(tags).RunWrite(a.conn)
		if err != nil || res.Inserted != len(user.Tags) {
			if res.Inserted > 0 {
				// Something went wrong, do best effort delete of inserted tags
				rdb.DB(a.dbName).Table("tagunique").GetAll(user.Tags).
					Filter(map[string]interface{}{"Source": user.Id}).Delete().RunWrite(a.conn)
			}
			return err, false
		}
	}

	_, err := rdb.DB(a.dbName).Table("users").Insert(&user).RunWrite(a.conn)
	if err != nil {
		return err, false
	}

	return nil, false
}

// 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
}

func main() {
	flag.Parse()

	session, err := r.Connect(r.ConnectOpts{
		Address:  *rethinkdbAddress,
		Database: *rethinkdbDatabase,
	})
	if err != nil {
		log.Fatal(err)
	}

	r.DB(*rethinkdbDatabase).TableCreate("scripts").Exec(session)
	r.DB(*rethinkdbDatabase).TableCreate("tokens").Exec(session)

	s := rpc.NewServer()
	s.RegisterCodec(json.NewCodec(), "application/json")
	s.RegisterService(&service.Service{
		Session: session,
	}, "Rexd")
	http.Handle("/rpc", s)

	http.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
		w.Write([]byte("lavab/rexd 0.1.0\n"))
	})

	log.Printf("Binding to %s", *bindAddress)
	if err := http.ListenAndServe(*bindAddress, nil); err != nil {
		log.Fatal(err)
	}
}

// TopicCreateP2P given two users creates a p2p topic
func (a *RethinkDbAdapter) TopicCreateP2P(initiator, invited *t.Subscription) error {
	initiator.Id = initiator.Topic + ":" + initiator.User
	// Don't care if the initiator changes own subscription
	_, err := rdb.DB(a.dbName).Table("subscriptions").Insert(initiator, rdb.InsertOpts{Conflict: "replace"}).
		RunWrite(a.conn)
	if err != nil {
		return err
	}

	// Ensure this is a new subscription. If one already exist, don't overwrite it
	invited.Id = invited.Topic + ":" + invited.User
	_, err = rdb.DB(a.dbName).Table("subscriptions").Insert(invited, rdb.InsertOpts{Conflict: "error"}).
		RunWrite(a.conn)
	if err != nil {
		// Is this a duplicate subscription? If so, ifnore it. Otherwise it's a genuine DB error
		if !rdb.IsConflictErr(err) {
			return err
		}
	}

	topic := &t.Topic{
		ObjHeader: t.ObjHeader{Id: initiator.Topic},
		Access:    t.DefaultAccess{Auth: t.ModeP2P, Anon: t.ModeBanned}}
	topic.ObjHeader.MergeTimes(&initiator.ObjHeader)
	return a.TopicCreate(topic)
}

func main() {

	config := a.NewConfig("config.gcfg")
	echo := echo.New()
	s := slack.New(config.Slack.Token)

	var session *r.Session

	session, err := r.Connect(r.ConnectOpts{
		Address:  fmt.Sprint(config.Database.URL, ":", config.Database.Port),
		Database: config.Database.Name,
		MaxIdle:  10,
		MaxOpen:  10,
	})
	if err != nil {
		log.Fatalln(err.Error())
	}
	session.SetMaxOpenConns(5)

	r.DBCreate(config.Database.Name).Exec(session)
	if err != nil {
		log.Println(err)
	}

	_, err = r.DB(config.Database.Name).TableCreate("quotes").RunWrite(session)
	if err != nil {
		fmt.Print(err)
	}
	_, err = r.DB(config.Database.Name).TableCreate("activities").RunWrite(session)
	if err != nil {
		fmt.Print(err)
	}

	// Middleware
	echo.Use(mw.Logger())
	echo.Use(mw.Recover())

	appcontext := &a.AppContext{
		Slack:  s,
		Config: config,
		Storage: &storage.Storage{
			Name:    "quotes",
			URL:     "192.168.10.10",
			Session: session,
		},
	}

	go appcontext.Monitor()

	//Routes
	a.Route(echo, appcontext)

	addr := fmt.Sprintf(":%d", config.App.Port)
	log.Printf("Starting server on: %s", addr)
	echo.Run(addr)
}

func dropDB() {
	var res *r.Cursor

	res, err = r.DB(dbName).TableList().ForEach(func(name r.Term) interface{} {
		r.DB(dbName).Table(name).Delete()
		return name
	}).Run(session)
	if err != nil {
		log.Println(res, err)
	}
}

// InitDBs prepares a RethinkDB instance to be used by the rethink server.
// Rethink servers will error if they are pointed at databases that haven't had InitDBs run on them.
// InitDBs is idempotent (unless rethink dies in the middle of the function)
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 {
		if _, ok := err.(gorethink.RQLRuntimeError); ok {
			return 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
			}
		}
	}
	if _, err := gorethink.DB(databaseName).Table(jobInfosTable).IndexCreateFunc(
		pipelineNameAndInputIndex,
		func(row gorethink.Term) interface{} {
			return []interface{}{
				row.Field("pipeline_name"),
				row.Field("input").Field("repo").Field("name"),
				row.Field("input").Field("id"),
			}
		}).RunWrite(session); err != nil {
		return err
	}
	if _, err := gorethink.DB(databaseName).Table(jobInfosTable).IndexCreateFunc(
		inputIndex,
		func(row gorethink.Term) interface{} {
			return []interface{}{
				row.Field("input").Field("repo").Field("name"),
				row.Field("input").Field("id"),
			}
		}).RunWrite(session); err != nil {
		return err
	}
	return nil
}

func setupDB(t *testing.T) (sess *r.Session, dbName string, err error) {

	dbName = fmt.Sprintf("db%d", int32(time.Now().Unix()))

	sess, err = r.Connect(r.ConnectOpts{
		Address:  "localhost:28015",
		Database: dbName,
	})
	if err != nil {
		return
	}

	_, err = r.DBCreate(dbName).Run(sess)
	if err != nil {
		t.Errorf("couldn't create table, err: %s ", err)
		return
	}
	r.DB(dbName).Wait().Run(sess)

	r.DB(dbName).TableCreate("test1", r.TableCreateOpts{PrimaryKey: "id"}).Exec(sess)
	r.DB(dbName).TableCreate("test2", r.TableCreateOpts{PrimaryKey: "id"}).Exec(sess)

	res, err := r.DB(dbName).TableList().Run(sess)
	if err != nil {
		t.Errorf("couldn't load table list, err: %s ", err)
		return
	}

	var tables []interface{}
	if err = res.All(&tables); err != nil {
		t.Errorf("couldn't load table list, err: %s ", err)
		return
	}

	if len(tables) != 2 {
		t.Errorf("table list off, %d    %v ", len(tables), tables)
		return
	}

	for idx, n := range names {

		var rec = struct {
			Name string
			Age  int
		}{Name: n, Age: 56 + idx}

		r.DB(dbName).Table("test1").Insert(rec).RunWrite(sess)
	}

	sess.Use(dbName)
	return
}

func checkTable(tname string) {
	session := connect()
	defer killSession(session)

	tables, err := r.DB(databaseName).TableList().Run(session)
	var result string
	for tables.Next(&result) {
		if result == tname {
			return
		}
	}
	_, err = r.DB(databaseName).TableCreate(tname).Run(session)
	LogError(err)
}

func (r *repo) InsertBlog(b *blogalert.Blog) error {
	_, err := gorethink.DB(Database).Table(BlogTable).
		Insert(newBlog(b)).RunWrite(r.session)

	r.cache.SetBlog(b)
	return err
}

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 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 SubscribeMessages(session *r.Session, roomId int, callback func(err error, message *Message)) (*r.Cursor, error) {
	cursor, err := r.DB(DB_NAME).Table("message").
		Between([]interface{}{roomId, r.MinVal}, []interface{}{roomId, r.MaxVal}, r.BetweenOpts{
			Index: "room_time",
		}).
		Changes().
		Run(session)

	if err != nil {
		return nil, err
	}

	go func() {
		var change struct {
			NewVal *Message `gorethink:"new_val"`
			OldVal *Message `gorethink:"old_val"`
		}
		for {
			ok := cursor.Next(&change)
			if !ok {
				err := cursor.Err()
				cursor.Close()
				callback(err, nil)
				return
			}

			if change.NewVal != nil {
				callback(nil, change.NewVal)
			}
		}
	}()

	return cursor, nil
}

func GetAuthorArticle(Author author) ([]Article, error) {
	res, err := r.DB(DB_NAME).Table(ARTICLE_TABLE).Run(session)
	if err != nil {
		return []Article{}, err
	}
	defer res.Close()
	var articles []Article
	var row interface{}
	for res.Next(&row) {
		var article Article
		if convert, ok := row.(map[string]interface{}); ok {
			article.title = convert["title"].(string)
			article.content = convert["content"].(string)
			article.id = convert["id"].(float64)
			if convert["author"] != nil {
				article.author = convert["author"].(float64)
			}
			if convert["tags"] != nil {
				tags := convert["tags"].([]interface{})
				for _, tag := range tags {
					article.tags = append(article.tags, tag.(float64))
				}
			}
			articles = append(articles, article)
		} else {
			fmt.Println("Cannot convert article row")
		}
	}
	return articles, nil
}

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.")
}

//SaveRelations saves a node to the database
func SaveRelations(relations []Relation) {
	session := connect()
	defer killSession(session)

	_, err := r.DB(databaseName).Table(relationTable).Insert(relations).RunWrite(session)
	LogError(err)
}

func addHandler(w http.ResponseWriter, req *http.Request) {
	input, err := ioutil.ReadAll(req.Body)
	defer req.Body.Close()

	var newRecipe Recipe

	err = json.Unmarshal(input, &newRecipe)
	if err != nil {
		log.Println(err.Error())
	}
	log.Println(newRecipe)

	resp, err := r.DB("recipes").Table("recipes").Insert(newRecipe).Run(session)
	if err != nil {
		log.Println(err.Error())
	}
	defer resp.Close()

	var response map[string]interface{}

	err = resp.One(&response)

	keys := response["generated_keys"].([]interface{})

	log.Println(keys)

	key := keys[0].(string)

	w.WriteHeader(http.StatusCreated)
	fmt.Fprintf(w, key)
}

func (s *Storage) DeleteActivity(id string) (*Activity, error) {

	rows, err := r.DB(s.Name).Table("activities").Get(id).Delete(r.DeleteOpts{ReturnChanges: true}).Run(s.Session)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var value r.WriteResponse
	rows.One(&value)

	var oldValueMap, ok = value.Changes[0].OldValue.(map[string]interface{})
	if !ok {
		fmt.Println("Type assertion failed :(")
	}
	fmt.Println("OldvalueMap: ", oldValueMap)

	var oldValueActivity Activity
	err = m.Decode(oldValueMap, &oldValueActivity)
	if err != nil {
		fmt.Println("err decoding: ", err)
	}

	fmt.Println("Oldvalueactivity: ", oldValueActivity)
	return &oldValueActivity, nil
}