func baseFind(o Option) r.Term {
	q := r.Table(resourceName)

	// slicing
	if o.Slice != "" {
		slice := strings.Split(o.Slice, ",")
		start, _ := strconv.Atoi(slice[0])
		end, _ := strconv.Atoi(slice[1])
		q = q.Slice(start, end)
	}

	// ordering
	if o.Order != "" {
		order := strings.Split(strings.ToLower(o.Order), ",")
		if len(order) == 2 && order[1] == "desc" {
			q = q.OrderBy(r.Desc(order[0]))
		} else {
			q = q.OrderBy(order[0])
		}
	}

	// filtering
	q = q.Filter(o.Filter)

	return q
}

func addOptions(q rdb.Term, value string, index string, opts *t.BrowseOpt) rdb.Term {
	var limit uint = 1024 // TODO(gene): pass into adapter as a config param
	var lower, upper interface{}

	if opts != nil {
		if opts.Since > 0 {
			lower = opts.Since
		} else {
			lower = rdb.MinVal
		}

		if opts.Before > 0 {
			upper = opts.Before
		} else {
			upper = rdb.MaxVal
		}

		if value != "" {
			lower = []interface{}{value, lower}
			upper = []interface{}{value, upper}
		}

		if opts.Limit > 0 && opts.Limit < limit {
			limit = opts.Limit
		}
	} else {
		lower = []interface{}{value, rdb.MinVal}
		upper = []interface{}{value, rdb.MaxVal}
	}

	return q.Between(lower, upper, rdb.BetweenOpts{Index: index}).
		OrderBy(rdb.OrderByOpts{Index: rdb.Desc(index)}).Limit(limit)
}

func loadLatestSales(websocketConnection *websocket.Conn) {
	results, err := r.Table("analytics").Filter(M{
		"type": "sale",
	}).OrderBy(
		r.Desc(r.Row.Field("data").Field("date")),
	).Limit(10).Map(func(row r.Term) interface{} {
		return row.Field("data")
	}).Run(rethinkSession)
	if err != nil {
		log.Printf("error loadLatestSales: %q", err)
	}
	defer results.Close()

	var latestSales []interface{}
	err = results.All(&latestSales)
	if err != nil {
		log.Println(err)
		return
	}

	log.Printf("sending latest %d sales\n", len(latestSales))

	socketOutgoingMessage := &SocketOutgoingMessage{
		Function: "loadLatestSales",
		Data: M{
			"type": "sale",
			"data": latestSales,
		},
	}

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

func Slot(r Repository) (*db.Cursor, error) {
	return db.Table(r.Table()).
		Filter(db.Row.Field("slot").Gt(0)).
		OrderBy(db.Desc("slot")).
		Pluck("id", "name", "slot").
		Run(r.Session())
}

func (r *repo) GetAllArticlesInBlog(blog *blogalert.Blog) ([]*blogalert.Article, error) {
	if blog == nil {
		return nil, nil
	}

	cursor, err := gorethink.DB(Database).Table(ArticleTable).
		OrderBy(gorethink.Desc("ts")).
		Filter(gorethink.Row.Field("blog").Eq(blog.URL.String())).
		Limit(100).
		Run(r.session)

	if err != nil {
		return nil, err
	}

	a := []*article{}

	err = cursor.All(&a)
	if err != nil {
		return nil, err
	}

	articles := make([]*blogalert.Article, 0, len(a))

	for _, v := range a {
		if article, err := v.ToArticle(blog); err == nil {
			articles = append(articles, article)
			r.cache.SetArticle(article)
		}
	}

	return articles, nil
}

func (r *repo) GetUserArticlesRead(UID string, blog *blogalert.Blog) ([]*blogalert.Article, error) {
	if blog == nil {
		return nil, nil
	}

	cursor, err := gorethink.DB(Database).Table(ArticleReadTable).
		OrderBy(gorethink.Desc("ts")).
		Filter(gorethink.Row.Field("uid").Eq(UID)).
		Filter(gorethink.Row.Field("blog").Eq(blog.URL.String())).
		Limit(100).
		Run(r.session)
	if err != nil {
		return nil, err
	}

	a := []*articleRead{}

	err = cursor.All(&a)
	if err != nil {
		return nil, err
	}

	articles := make([]*blogalert.Article, 0, len(a))

	for _, v := range a {
		if article, err := r.GetArticle(v.ArticleURL); err == nil {
			articles = append(articles, article)
		}
	}

	return articles, nil
}

func List(r Repository, nameField string, offset int, limit int) (*db.Cursor, error) {
	return db.Table(r.Table()).
		OrderBy(db.OrderByOpts{Index: db.Desc("id")}).
		Slice(offset, limit).
		Pluck("id", nameField).
		Run(r.Session())
}

func subscribeChannelMessage(client *Client, data interface{}) {
	go func() {
		eventData := data.(map[string]interface{})
		val, ok := eventData["channelId"]
		if !ok {
			return
		}
		channelId, ok := val.(string)
		if !ok {
			return
		}
		stop := client.NewStopChannel(MessageStop)
		cursor, err := r.Table("message").
			OrderBy(r.OrderByOpts{Index: r.Desc("createdAt")}).
			Filter(r.Row.Field("channelId").Eq(channelId)).
			Changes(r.ChangesOpts{IncludeInitial: true}).
			Run(client.session)

		if err != nil {
			client.send <- Message{"error", err.Error()}
			return
		}
		changeFeedHelper(cursor, "message", client.send, stop)
	}()
}

func (m *Manager) Events(limit int) ([]*shipyard.Event, error) {
	res, err := r.Table(tblNameEvents).OrderBy(r.Desc("Time")).Limit(limit).Run(m.session)
	if err != nil {
		return nil, err
	}
	var events []*shipyard.Event
	if err := res.All(&events); err != nil {
		return nil, err
	}
	return events, nil
}

func main() {
	log.Println("VERSION: ", VERSION)
	e := echo.New()
	e.Use(middleware.Logger())
	e.Use(middleware.Recover())
	e.Use(cors.Default().Handler)
	e.Index("public/index.html")
	e.Static("/", "public")

	if len(os.Args) < 2 {
		fmt.Println("\nPlease input callback url.\n  Usage: app.exe [callback_url]")
		os.Exit(0)
	}
	callback_url = os.Args[1]
	//e.Favicon("public/favicon.ico")

	server, err := socketio.NewServer(nil)
	if err != nil {
		log.Fatal(err)
	}
	e.Get("/publish/photo", CallbackHandler)
	e.Post("/publish/photo", ReceiveHandler)
	e.Get("/socket.io/", server)

	server.On("connection", func(so socketio.Socket) {
		log.Println("====================== on connection ======================")
		result := make([]interface{}, 12)
		cur, err := r.Table("instacat").OrderBy(r.OrderByOpts{
			Index: r.Desc("time"),
		}).Limit(12).Run(session)
		if err != nil {
			log.Println(err.Error())
		}
		cur.All(&result)
		fmt.Println("Get result over. ")
		err = so.Emit("recent", result)
		if err != nil {
			log.Println(err.Error())
		}
		so.On("disconnect", func() {
			log.Println("on disconnect")
		})

		RealtimeChangefeed(so)
	})

	//go SubscribeTag()
	go sub()
	port := ":3000"
	log.Printf("Starting HTTP service on %s ...", port)
	e.Run(port)

}

func (m DefaultManager) Events(limit int) ([]*shipyard.Event, error) {
	t := r.Table(tblNameEvents).OrderBy(r.Desc("Time"))
	if limit > -1 {
		t.Limit(limit)
	}
	res, err := t.Run(m.session)
	if err != nil {
		return nil, err
	}
	events := []*shipyard.Event{}
	if err := res.All(&events); err != nil {
		return nil, err
	}
	return events, nil
}

func (s *Storage) GetLatestQuote() Quote {

	rows, err := r.Table("quotes").OrderBy(r.Desc("timestamp")).Run(s.Session)
	if err != nil {
		fmt.Println(err)
	}
	defer rows.Close()

	var quote Quote
	err2 := rows.One(&quote)
	if err2 != nil {
		fmt.Println(err2)
	}

	fmt.Printf("Fetch one record %#v\n", quote)

	return quote
}

func (c *rethinkClient) GetPipelineRunStatusLatest(id string) (*pps.PipelineRunStatus, error) {
	cursor, err := c.statuses.
		GetAllByIndex("pipeline_run_id", id).
		OrderBy(gorethink.Desc("timestamp")).
		Nth(0).
		ToJSON().
		Run(c.session)
	if err != nil {
		return nil, err
	}
	data := ""
	if !cursor.Next(&data) {
		return nil, cursor.Err()
	}
	var pipelineRunStatus pps.PipelineRunStatus
	if err := jsonpb.UnmarshalString(data, &pipelineRunStatus); err != nil {
		return nil, err
	}
	return &pipelineRunStatus, nil
}

func (a *rethinkAPIServer) ListPipelineInfos(ctx context.Context, request *google_protobuf.Empty) (response *persist.PipelineInfos, err error) {
	defer func(start time.Time) { a.Log(request, response, err, time.Since(start)) }(time.Now())
	pipelineInfoObjs, err := a.getAllMessages(
		pipelineInfosTable,
		func() proto.Message { return &persist.PipelineInfo{} },
		func(term gorethink.Term) gorethink.Term {
			return term.OrderBy(gorethink.Desc("created_at"))
		},
	)
	if err != nil {
		return nil, err
	}
	pipelineInfos := make([]*persist.PipelineInfo, len(pipelineInfoObjs))
	for i, pipelineInfoObj := range pipelineInfoObjs {
		pipelineInfos[i] = pipelineInfoObj.(*persist.PipelineInfo)
	}
	return &persist.PipelineInfos{
		PipelineInfo: pipelineInfos,
	}, nil
}

func (c *Connection) GetBookmarks(userID string, page int64) ([]Bookmark, error) {
	var bookmarks []Bookmark

	cursor, err := r.DB("magnet").
		Table("bookmarks").
		OrderBy(r.OrderByOpts{r.Desc("Created")}).
		Filter(r.Row.Field("User").Eq(userID)).
		Skip(50 * page).
		Limit(50).
		Run(c.session)

	if err != nil {
		log.Print(err)
		return bookmarks, err
	}

	cursor.All(&bookmarks)
	cursor.Close()
	return bookmarks, err
}

func (c *rethinkClient) GetAllPipelineRunStatuses(pipelineRunID string) ([]*pps.PipelineRunStatus, error) {
	cursor, err := c.pipelineRunStatuses.
		GetAllByIndex("pipeline_run_id", pipelineRunID).
		OrderBy(gorethink.Desc("timestamp")).
		Without("id").
		Map(rethinkToJSON).
		Run(c.session)
	if err != nil {
		return nil, err
	}
	var pipelineRunStatuses []*pps.PipelineRunStatus
	data := ""
	for cursor.Next(&data) {
		var pipelineRunStatus pps.PipelineRunStatus
		if err := jsonpb.UnmarshalString(data, &pipelineRunStatus); err != nil {
			return nil, err
		}
		pipelineRunStatuses = append(pipelineRunStatuses, &pipelineRunStatus)
	}
	return pipelineRunStatuses, cursor.Err()
}

func orders(w http.ResponseWriter, q *http.Request) {
	ctx, err := NewContext(q)
	if err != nil {
		log.Println(err)
	}

	rows, err := r.Db("goShop").Table("orders").OrderBy(r.Desc("Date")).Run(ctx.Db)

	if err != nil {
		log.Println(err)
	}
	var t []Order
	err = rows.All(&t)
	if err != nil {
		log.Println(err)
	}

	newmap := map[string]interface{}{"Orders": t}

	templatesA.ExecuteTemplate(w, "orders", newmap)

}

func (a *rethinkAPIServer) GetJobInfosByPipeline(ctx context.Context, request *pps.Pipeline) (response *persist.JobInfos, err error) {
	defer func(start time.Time) { a.Log(request, response, err, time.Since(start)) }(time.Now())
	jobInfoObjs, err := a.getMessagesByIndex(
		jobInfosTable,
		pipelineNameIndex,
		request.Name,
		func() proto.Message { return &persist.JobInfo{} },
		func(term gorethink.Term) gorethink.Term {
			return term.OrderBy(gorethink.Desc("created_at"))
		},
	)
	if err != nil {
		return nil, err
	}
	jobInfos := make([]*persist.JobInfo, len(jobInfoObjs))
	for i, jobInfoObj := range jobInfoObjs {
		jobInfos[i] = jobInfoObj.(*persist.JobInfo)
	}
	return &persist.JobInfos{
		JobInfo: jobInfos,
	}, nil
}

// ordered by time, latest to earliest
func (a *rethinkAPIServer) GetJobLogs(ctx context.Context, request *pps.Job) (response *persist.JobLogs, err error) {
	defer func(start time.Time) { a.Log(request, response, err, time.Since(start)) }(time.Now())
	jobLogObjs, err := a.getMessagesByIndex(
		jobLogsTable,
		jobIDIndex,
		request.Id,
		func() proto.Message { return &persist.JobLog{} },
		func(term gorethink.Term) gorethink.Term {
			return term.OrderBy(gorethink.Desc("timestamp"))
		},
	)
	if err != nil {
		return nil, err
	}
	jobLogs := make([]*persist.JobLog, len(jobLogObjs))
	for i, jobLogObj := range jobLogObjs {
		jobLogs[i] = jobLogObj.(*persist.JobLog)
	}
	return &persist.JobLogs{
		JobLog: jobLogs,
	}, nil
}

func (c *Connection) Search(userID string, params martini.Params, query string) ([]interface{}, error) {
	var response []interface{}
	page, _ := strconv.ParseInt(params["page"], 10, 16)

	cursor, err := r.DB("magnet").
		Table("bookmarks").
		OrderBy(r.OrderByOpts{r.Desc("Created")}).
		Filter(r.Row.Field("Title").Match("(?i)" + query).
			And(r.Row.Field("User").Eq(userID))).
		Skip(50 * page).
		Limit(50).
		Run(c.session)

	if err != nil {
		log.Print(err)
		return nil, err
	}

	cursor.All(&response)
	cursor.Close()
	return response, err
}