// Test a Cassandra database
func TestCassandra(t *testing.T) {
	cluster := gocql.NewCluster("127.0.0.1")
	cluster.Keyspace = "testkeyspace"
	cluster.Consistency = gocql.Quorum
	session, err := cluster.CreateSession()
	if err != nil {
		t.Fatal("createsession", err)
	}
	defer session.Close()

	err = session.Query(`CREATE KEYSPACE IF NOT EXISTS testkeyspace with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };`).Exec()
	if err != nil {
		t.Fatal("createkeyspace", err)
	}
	_ = session.Query(`DROP TABLE IF EXISTS passwords`).Exec()

	err = session.Query(`CREATE TABLE testkeyspace.passwords(password text, PRIMARY KEY(password));`).Exec()
	if err != nil {
		t.Fatal("createkeyspace", err)
	}

	defer session.Query(`DROP TABLE IF EXISTS passwords`).Exec()

	db := New(session, "passwords")
	err = drivers.TestImport(db)
	if err != nil {
		t.Fatal(err)
	}

	err = drivers.TestData(db)
	if err != nil {
		t.Fatal(err)
	}
}

func TestCreate(t *testing.T) {
	cluster := gocql.NewCluster("127.0.0.1")
	cluster.ProtoVersion = 4
	cluster.Keyspace = "aaas"
	session, err := cluster.CreateSession()
	if err != nil {
		panic(err)
	}
	defer session.Close()

	asset := &Asset{
		Name:        "test1",
		Path:        []string{"a", "b"},
		ContentType: "image/jpeg",
	}
	err = asset.Save(session)
	if err != nil {
		log.Panic(err)
	}

	err = asset.Delete(session, asset.Id.String())
	if err != nil {
		log.Panic(err)
	}
	log.Println(asset)
}

// NewAPI creates a new instance of API from the given configuration.
func NewAPI(config api.Config) (api.API, error) {
	file, err := os.Open(config.ConversionRulesPath)
	if err != nil {
		return nil, err
	}
	bytes, err := ioutil.ReadAll(file)
	if err != nil {
		return nil, err
	}
	ruleset, err := LoadYAML(bytes)
	if err != nil {
		return nil, err
	}
	clusterConfig := gocql.NewCluster()
	clusterConfig.Hosts = config.Hosts
	clusterConfig.Keyspace = config.Keyspace
	clusterConfig.Timeout = time.Second * 30
	db, err := NewCassandraDatabase(clusterConfig)
	if err != nil {
		return nil, err
	}
	return &defaultAPI{
		db:      db,
		ruleset: ruleset,
	}, nil
}

func newCassandraDriver(ctx *cli.Context) (drivers.Driver, error) {

	urls := strings.Split(ctx.String("cassandra-hosts"), ",")

	cluster := gocql.NewCluster()
	cluster.Hosts = urls
	cluster.Port = ctx.Int("cassandra-port")
	cluster.Keyspace = ctx.String("cassandra-keyspace")
	cluster.Consistency = gocql.Quorum
	session, err := cluster.CreateSession()

	if err != nil {
		log.Errorf("cassandra.go: Error while creating cassandra cluster: hosts:%v err:%v", cluster.Hosts, err)
		return nil, err
	}

	log.Infoln("cassandra.go: connected to cassandra cluster")

	d := &CassandraDriver{
		Session: session,
		Cluster: cluster,
	}

	return d, nil
}

func main() {
	// connect to the cluster
	cluster := gocql.NewCluster("192.168.1.54")
	cluster.Keyspace = "example"
	cluster.Consistency = gocql.Quorum
	session, _ := cluster.CreateSession()
	defer session.Close()

	// insert a tweet
	// if err := session.Query(`INSERT INTO tweet (timeline, id, text, name, time) VALUES (?, ?, ?, ?, ?)`,
	// 	"me", gocql.TimeUUID(), "hello world", "Ninoy", time.Now()).Exec(); err != nil {
	// 	log.Fatal(err)
	// }

	var id gocql.UUID
	var text string

	/* Search for a specific set of records whose 'timeline' column matches
	 * the value 'me'. The secondary index that we created earlier will be
	 * used for optimizing the search */
	// if err := session.Query(`SELECT id, text FROM tweet WHERE timeline = ? LIMIT 1`,
	// 	"me").Scan(&id, &text); err != nil {
	// 	log.Fatal(err)
	// }
	// fmt.Println("Tweet:", id, text)

	// list all tweets
	iter := session.Query(`SELECT id, text FROM tweet WHERE time < ? allow filtering`, time.Now()).Iter()
	for iter.Scan(&id, &text) {
		fmt.Println("Tweet:", id, text)
	}
	if err := iter.Close(); err != nil {
		log.Fatal(err)
	}
}

//getTracingLog gets the tracing log by scanning the cql database
func getTracingLog() string {
	// connect to the cluster
	cluster := gocql.NewCluster(config.Cassandra.IP)
	cluster.Keyspace = "system_traces"
	cluster.Consistency = gocql.Quorum
	session, err := cluster.CreateSession()
	if err != nil {
		panic(fmt.Sprintf("error creating session: %v", err))
	}
	defer session.Close()

	e := events{}

	var queryEvents = `SELECT session_id, event_id, activity, source, source_elapsed, thread FROM system_traces.events;`
	iter := session.Query(queryEvents).Consistency(gocql.One).Iter()

	fmt.Println("Fetched events")
	var eOutput = ""
	for iter.Scan(&e.SessionID, &e.EventID, &e.Activity, &e.Source, &e.SourceElapsed, &e.Thread) {
		fmt.Println("Event: ", e.EventID, e.Activity, e.Source, e.SourceElapsed, e.Thread)
		eOutput += "Event: " + "\t" + e.EventID + "\t" + e.Activity + "\t" + e.Source + "\t" + e.SourceElapsed + "\t" + e.Thread + "\n"
	}
	if err := iter.Close(); err != nil {
		log.Fatal(err)
	}

	return eOutput
}

func init() {
	kname := "test_ihopeudonthaveakeyspacenamedlikedthis"
	var err error

	cluster := gocql.NewCluster(getTestHosts()...)
	cluster.Consistency = gocql.One
	cluster.Timeout = 10 * time.Second               // Travis' C* is sloooow
	cluster.MaxWaitSchemaAgreement = 2 * time.Minute // travis might be slow
	cluster.RetryPolicy = &gocql.SimpleRetryPolicy{
		NumRetries: 3}
	sess, err := cluster.CreateSession()
	if err != nil {
		panic(err)
	}
	conn := &connection{q: goCQLBackend{session: sess}}
	ns = conn.KeySpace(kname)

	err = conn.DropKeySpace(kname)
	if err != nil {
		panic(err)
	}
	err = conn.CreateKeySpace(kname)
	if err != nil {
		panic(err)
	}
}

// New db adaptor
func NewAdaptor(urls []string, username, password string) *Adaptor {
	cluster := gocql.NewCluster(urls...)
	cluster.Keyspace = "events"
	cluster.ProtoVersion = 3

	// cluster.Authenticator = gocql.PasswordAuthenticator{
	// 	Username: username,
	// 	Password: password,
	// }

	session, err := cluster.CreateSession()
	if err != nil {
		log.Fatalf("Cassandra conn error: %s", err.Error())
	}
	qe := gocassa.GoCQLSessionToQueryExecutor(session)
	conn := gocassa.NewConnection(qe)
	gocassaSession := conn.KeySpace("events")

	return &Adaptor{
		Session:            gocassaSession,
		urls:               urls,
		eventTimeDateRange: 7 * 24 * time.Hour,
		Username:           username,
		Password:           password,
	}
}

func init() {
	kname := "test_ihopeudonthaveakeyspacenamedlikedthis"
	var err error

	c, err := Connect(getTestHosts(), "", "")
	if err != nil {
		panic(err)
	}
	err = c.DropKeySpace(kname)
	if err != nil {
		panic(err)
	}
	err = c.CreateKeySpace(kname)
	if err != nil {
		panic(err)
	}

	cluster := gocql.NewCluster(getTestHosts()...)
	cluster.Consistency = gocql.One
	cluster.Timeout = 1500 * time.Millisecond // Travis' C* is sloooow
	cluster.RetryPolicy = &gocql.SimpleRetryPolicy{
		NumRetries: 3}
	sess, err := cluster.CreateSession()
	if err != nil {
		panic(err)
	}
	conn := &connection{q: goCQLBackend{session: sess}}
	ns = conn.KeySpace(kname)
}

// NewServer returns a new instance of Server built from a config.
func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {

	elasticsearch := elastigo.NewConn()
	elasticsearch.SetFromUrl(c.Meta.ElasticSearchUrl)

	db := gocql.NewCluster(c.Database.CassandraNodes...)
	db.Keyspace = c.Database.CassandraKeyspace
	db.NumConns = c.Database.CassandraConns
	db.Discovery = gocql.DiscoveryConfig{
		DcFilter:   "",
		RackFilter: "",
		Sleep:      30 * time.Second,
	}

	s := &Server{
		buildInfo: *buildInfo,
		err:       make(chan error),
		closing:   make(chan struct{}),

		Hostname:      c.Meta.Hostname,
		BindAddress:   c.Meta.BindAddress,
		elasticsearch: elasticsearch,
		cassandra:     db,
	}

	// Append services.
	//s.appendMongoService(c.Mongo)

	s.appendMetricsReportingService(c.Meta)
	s.appendHTTPDService(c.HTTPD)
	s.appendRegistrationService(c.Registration, c.RegMeta)
	return s, nil
}

func main() {
	// set the parameters for our cassandra connection
	cluster := gocql.NewCluster("127.0.0.1")
	cluster.DiscoverHosts = true
	cluster.DefaultTimestamp = false

	// establish the cassandra session
	session, err := cluster.CreateSession()
	if err != nil {
		fmt.Println("could not create session")
		os.Exit(1)
	}
	defer session.Close()

	// get a list of iles in the logs folder
	files, err := ioutil.ReadDir(path.Join("..", logDir))
	if err != nil {
		fmt.Println("could not read logs folder")
		os.Exit(1)
	}

	// iterate through files for processing
	for _, file := range files {
		err := process(file)
		if err != nil {
			fmt.Printf("%v", err)
		}
	}

	fmt.Printf("processed %v files\n", len(files))
	os.Exit(0)
}

func main() {
	log.Println("Main")
	db := gocql.NewCluster("127.0.0.1", "127.0.0.2", "127.0.0.3")
	db.Keyspace = "test"
	db.ProtoVersion = 4
	db.Consistency = gocql.Quorum

	log.Println(db)

	now := time.Now()

	//numberOfWorker := 1
	//w := Writer{100000, db}
	//w.Run()

	numberOfWorker := 3
	var wg sync.WaitGroup
	wg.Add(numberOfWorker)
	for i := 0; i < numberOfWorker; i++ {
		go func(i int, conn *gocql.ClusterConfig, wg *sync.WaitGroup) {
			w := Writer{100000, conn}
			w.Run()
			log.Println("Writer", i, "finished")
			wg.Done()
		}(i, db, &wg)

	}

	wg.Wait()

	log.Println(numberOfWorker, "Writer", "took", time.Now().Sub(now))

}

//getSessionLog returns the cassandra session log by scanning the database
func getSessionLog() string {

	// connect to the cluster
	cluster := gocql.NewCluster(config.Cassandra.IP)
	cluster.Keyspace = "system_traces"
	cluster.Consistency = gocql.Quorum
	session, err := cluster.CreateSession()
	if err != nil {
		panic(fmt.Sprintf("error creating session: %v", err))
	}
	defer session.Close()

	s := sessions{}

	var querySessions = `SELECT session_id, coordinator, duration, parameters, request, started_at FROM system_traces.sessions;`
	iter := session.Query(querySessions).Consistency(gocql.One).Iter()

	fmt.Println("Fetched sessions")
	var sOutput = ""
	for iter.Scan(&s.SessionID, &s.Coordinator, &s.Duration, &s.Parameters, &s.Request, &s.StartedAt) {
		//fmt.Println("Session: ", &s.SessionID, s.Coordinator, s.Duration, s.Parameters, s.Request, s.StartedAt)
		var param = "{"
		for paramKeys, paramValue := range s.Parameters {
			param = param + paramKeys + ": " + paramValue + ", "
		}

		param += "} "
		sOutput += "Session:\t" + s.SessionID + "\t" + s.Coordinator + "\t" + s.Duration + "\t" + param + "\t" + s.Request + "\t" + s.StartedAt.String() + "\n"
	}
	if err := iter.Close(); err != nil {
		log.Fatal(err)
	}
	log.Println(sOutput)
	return sOutput
}

// Cassandra Driver URL format:
// cassandra://host:port/keyspace
//
// Example:
// cassandra://localhost/SpaceOfKeys
func (driver *Driver) Initialize(rawurl string) error {
	u, err := url.Parse(rawurl)

	cluster := gocql.NewCluster(u.Host)
	cluster.Keyspace = u.Path[1:len(u.Path)]
	cluster.Consistency = gocql.All
	cluster.Timeout = 1 * time.Minute

	// Check if url user struct is null
	if u.User != nil {
		password, passwordSet := u.User.Password()

		if passwordSet == false {
			return fmt.Errorf("Missing password. Please provide password.")
		}

		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username: u.User.Username(),
			Password: password,
		}

	}

	driver.session, err = cluster.CreateSession()

	if err != nil {
		return err
	}

	if err := driver.ensureVersionTableExists(); err != nil {
		return err
	}
	return nil
}

func main() {
	cluster := gocql.NewCluster("127.0.0.1")

	for v := 3; v < 4; v++ {
		cluster.ProtoVersion = v
		session, err := cluster.CreateSession()
		if err != nil {
			log.Printf("Error openning session: %s\n", err.Error())
			continue
		}

		session.SetPageSize(123)
		session.SetConsistency(gocql.LocalQuorum)

		qry := session.Query(`
			CREATE KEYSPACE foo 
			WITH REPLICATION {
				'class': 'SimpleStrategy', 
				'replication_factor': 3
			}`,
			"foo",
			"bar",
		)

		// qry.Consistency(gocql.Any).PageSize(987)
		qry.SerialConsistency(gocql.LocalSerial)
		if err := qry.Exec(); err != nil {
			log.Printf("Error executing query: %s\n", err.Error())
		}

		session.Close()
	}
}

func (o *Oinker) NewGraph() inject.Graph {
	graph := inject.NewGraph()

	var server *http.ServeMux
	graph.Define(&server, inject.NewProvider(http.NewServeMux))

	var cqlCluster *gocql.ClusterConfig
	graph.Define(&cqlCluster, inject.NewProvider(func() *gocql.ClusterConfig {
		//TODO: use DiscoverHosts?
		return gocql.NewCluster(o.CQLHosts...)
	}))

	var oinkRepo model.OinkRepo
	if len(o.CQLHosts) > 0 {
		graph.Define(&oinkRepo, inject.NewProvider(model.NewCQLOinkRepo, &cqlCluster))
	} else {
		graph.Define(&oinkRepo, inject.NewProvider(model.NewMockOinkRepo))
	}

	var assetsController *controller.AssetsController
	graph.Define(&assetsController, inject.NewProvider(controller.NewAssetsController))

	var indexController *controller.IndexController
	graph.Define(&indexController, inject.NewProvider(controller.NewIndexController, &oinkRepo))

	var oinkController *controller.OinkController
	graph.Define(&oinkController, inject.NewProvider(controller.NewOinkController, &oinkRepo))

	var analyticsController *controller.AnalyticsController
	graph.Define(&analyticsController, inject.NewProvider(controller.NewAnalyticsController, &oinkRepo))

	return graph
}

// OpenCassandraContainerConnection ...
func OpenCassandraContainerConnection(tries int, delay time.Duration) (c ContainerID, session *gocql.Session, err error) {
	c, ip, port, err := SetupCassandraContainer()
	if err != nil {
		return c, nil, fmt.Errorf("Could not set up Cassandra container: %v", err)
	}

	for try := 0; try <= tries; try++ {
		time.Sleep(delay)
		node := fmt.Sprintf("%s:%d", ip, port)
		log.Printf("Try %d: Connecting %s", try, node)

		cluster := gocql.NewCluster(ip)
		cluster.Keyspace = "system"
		cluster.Port = port
		cluster.Consistency = gocql.Quorum
		session, _ := cluster.CreateSession()
		if err := session.Query(`SELECT COUNT(*) FROM system.local`).Exec(); err == nil {
			log.Printf("Try %d: Successfully connected to %v", try, ip)
			return c, session, nil
		}
		log.Printf("Try %d: Could not set up Cassandra container: %v", try, err)

	}
	return c, nil, errors.New("Could not set up Cassandra container.")
}

func main() {
	runtime.GOMAXPROCS(runtime.NumCPU())
	// defer client.Close()
	// connect to the cluster
	cluster := gocql.NewCluster("172.31.53.229")
	cluster.Keyspace = "example"
	//cluster.Consistency = gocql.Quorum
	session, _ = cluster.CreateSession()

	if err := session.Query(`TRUNCATE totals`).Exec(); err != nil {
		log.Fatal(err)
	}

	ch = make(chan Event, 1000)
	time.AfterFunc(1*time.Second, func() {
		for i := 1; i < 72; i++ {
			fmt.Println("starting worker %d", i)
			go worker(ch)
		}
	})

	http.HandleFunc("/vote", voteHandler)
	http.HandleFunc("/loaderio-35df9c4fffde902e3b0e3e0115816d82.html", validationHandler)
	http.ListenAndServe(":80", nil)
}

func processCQLfile(filename string) {
	//restore te proper seperators to the file name
	filename = strings.Replace(filename, "|", "/", -1)

	cluster := gocql.NewCluster(config.DbHost)
	cluster.Keyspace = "joulepersecond"
	cluster.Consistency = gocql.Quorum
	session, _ := cluster.CreateSession()
	defer session.Close()

	file, err := os.Open(filename)
	if err != nil {
		log.Printf("Location:%v", err)
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)

	for scanner.Scan() {
		cql := scanner.Text()
		//insert each line of the file
		if err := session.Query(cql).Exec(); err != nil {
			log.Printf("Location:%v", err)
		}
	}
}

// InitDB returns a valid database object and creates tables if necessary
func InitDB(config *DBConfig) (*gocql.ClusterConfig, error) {
	c := gocql.NewCluster(config.DBHosts...)
	s, err := c.CreateSession()
	if err != nil {
		return nil, err
	}
	err = createKeyspace(s, config.Keyspace)
	if err != nil {
		return nil, err
	}
	s.Close()
	c.Keyspace = config.Keyspace
	c.Consistency = config.Consistency
	s, err = c.CreateSession() //new session w/ keyspace set
	if err != nil {
		return nil, err
	}
	defer s.Close()
	err = createUDTs(s)
	if err != nil {
		return nil, err
	}
	err = createTables(s)
	if err != nil {
		return nil, err
	}
	return c, nil
}