// auth for goim handshake with client, use rsa & aes.
func (server *Server) auth(rr io.Reader, wr io.Writer, fr Flusher, dbm cipher.BlockMode, p *Proto) (subKey string, heartbeat time.Duration, err error) {
	log.Debug("get auth request protocol")
	if err = server.readRequest(rr, p); err != nil {
		return
	}
	if p.Operation != OP_AUTH {
		log.Warn("auth operation not valid: %d", p.Operation)
		err = ErrOperation
		return
	}
	if p.Body, err = server.cryptor.Decrypt(dbm, p.Body); err != nil {
		log.Error("auth decrypt client proto error(%v)", err)
		return
	}
	if subKey, heartbeat, err = server.operator.Connect(p); err != nil {
		log.Error("operator.Connect error(%v)", err)
		return
	}
	log.Debug("send auth response protocol")
	p.Body = nil
	p.Operation = OP_AUTH_REPLY
	if err = server.sendResponse(wr, fr, p); err != nil {
		log.Error("[%s] server.SendResponse() error(%v)", subKey, err)
	}
	return
}

/*
 *	GameStateFinishing
 *	All bets are over. Now collect the words made by users and compute the winner and give him the pot amount.
 *	BREAKS ON: computing winner finished
 */
func (g *Game) GameStateFinishing() {
	defer func() {
		recover()
	}()
	for {
		select {
		case <-g.stateFinishing:
			log.Debug("STATE: finishing")
			winnerIds := make([]string, 0)
			winners := g.computeWinner()
			if len(winners) > 0 {
				winnerShare := g.PotAmount / len(winners)
				for _, p := range winners {
					if p != nil {
						p.Cash += winnerShare
						winnerIds = append(winnerIds, strconv.Itoa(p.Id))
					}
				}
				log.Debug("Winners: %v", winnerIds)
			} else {
				log.Debug("No winner. House wins the pot.")
			}
			g.sendUpdate(ACT_GAME_OVER, strings.Join(winnerIds, ","), 0, nil)

			time.Sleep(time.Millisecond * DUR_WAIT_GAME_OVER)

			log.Debug("---- end of game ----")

			g.State = GS_WAITING
			g.stateWaiting <- 1
		}
	}
}

func (self *HttpServer) createDbUser(w libhttp.ResponseWriter, r *libhttp.Request) {
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		w.WriteHeader(libhttp.StatusInternalServerError)
		w.Write([]byte(err.Error()))
		return
	}

	newUser := &NewUser{}
	err = json.Unmarshal(body, newUser)
	if err != nil {
		w.WriteHeader(libhttp.StatusBadRequest)
		w.Write([]byte(err.Error()))
		return
	}

	db := r.URL.Query().Get(":db")

	self.tryAsDbUserAndClusterAdmin(w, r, func(u User) (int, interface{}) {
		username := newUser.Name
		if err := self.userManager.CreateDbUser(u, db, username, newUser.Password); err != nil {
			log.Error("Cannot create user: %s", err)
			return errorToStatusCode(err), err.Error()
		}
		log.Debug("Created user %s", username)
		if newUser.IsAdmin {
			err = self.userManager.SetDbAdmin(u, db, newUser.Name, true)
			if err != nil {
				return libhttp.StatusInternalServerError, err.Error()
			}
		}
		log.Debug("Successfully changed %s password", username)
		return libhttp.StatusOK, nil
	})
}

func (self *WAL) processAppendEntry(e *appendEntry) {
	nextRequestNumber := self.state.getNextRequestNumber()
	e.request.RequestNumber = proto.Uint32(nextRequestNumber)

	if len(self.logFiles) == 0 {
		if _, err := self.createNewLog(nextRequestNumber); err != nil {
			e.confirmation <- &confirmation{0, err}
			return
		}
		self.state.FirstSuffix = nextRequestNumber
	}

	lastLogFile := self.logFiles[len(self.logFiles)-1]
	self.assignSequenceNumbers(e.shardId, e.request)
	logger.Debug("appending request %d", e.request.GetRequestNumber())
	err := lastLogFile.appendRequest(e.request, e.shardId)
	if err != nil {
		e.confirmation <- &confirmation{0, err}
		return
	}
	self.state.CurrentFileOffset = self.logFiles[len(self.logFiles)-1].offset()

	self.requestsSinceLastIndex++
	self.requestsSinceLastBookmark++
	self.requestsSinceLastFlush++
	self.requestsSinceRotation++
	logger.Debug("requestsSinceRotation: %d", self.requestsSinceRotation)
	if rotated, err := self.rotateTheLogFile(nextRequestNumber); err != nil || rotated {
		e.confirmation <- &confirmation{e.request.GetRequestNumber(), err}
		return
	}

	self.conditionalBookmarkAndIndex()
	e.confirmation <- &confirmation{e.request.GetRequestNumber(), nil}
}

func (self *Passthrough) Yield(seriesIncoming *protocol.Series) (bool, error) {
	log.Debug("PassthroughEngine YieldSeries %d", len(seriesIncoming.Points))

	self.limiter.calculateLimitAndSlicePoints(seriesIncoming)
	if len(seriesIncoming.Points) == 0 {
		return false, nil
	}

	if self.series == nil {
		self.series = seriesIncoming
	} else if self.series.GetName() != seriesIncoming.GetName() {
		log.Debug("Yielding to %s: %s", self.next.Name(), self.series)
		ok, err := self.next.Yield(self.series)
		if !ok || err != nil {
			return ok, err
		}
		self.series = seriesIncoming
	} else if len(self.series.Points) > self.maxPointsInResponse {
		log.Debug("Yielding to %s: %s", self.next.Name(), self.series)
		ok, err := self.next.Yield(self.series)
		if !ok || err != nil {
			return ok, err
		}
		self.series = seriesIncoming
	} else {
		self.series = common.MergeSeries(self.series, seriesIncoming)
	}
	return !self.limiter.hitLimit(seriesIncoming.GetName()), nil
}

func (c *DefaultServerCodec) ReadRequestHeader(rd *bufio.Reader, proto *Proto) (err error) {
	if err = binary.Read(rd, binary.BigEndian, &proto.PackLen); err != nil {
		log.Error("packLen: binary.Read() error(%v)", err)
		return
	}
	log.Debug("packLen: %d", proto.PackLen)
	if proto.PackLen > maxPackLen {
		return ErrProtoPackLen
	}
	if err = binary.Read(rd, binary.BigEndian, &proto.HeaderLen); err != nil {
		log.Error("headerLen: binary.Read() error(%v)", err)
		return
	}
	log.Debug("headerLen: %d", proto.HeaderLen)
	if proto.HeaderLen != rawHeaderLen {
		return ErrProtoHeaderLen
	}
	if err = binary.Read(rd, binary.BigEndian, &proto.Ver); err != nil {
		log.Error("protoVer: binary.Read() error(%v)", err)
		return
	}
	log.Debug("protoVer: %d", proto.Ver)
	if err = binary.Read(rd, binary.BigEndian, &proto.Operation); err != nil {
		log.Error("Operation: binary.Read() error(%v)", err)
		return
	}
	log.Debug("operation: %d", proto.Operation)
	if err = binary.Read(rd, binary.BigEndian, &proto.SeqId); err != nil {
		log.Error("seqId: binary.Read() error(%v)", err)
		return
	}
	log.Debug("seqId: %d", proto.SeqId)
	return
}

func ReadProto(rd *bufio.Reader, proto *Proto) (err error) {
	// read
	if err = binary.Read(rd, binary.BigEndian, &proto.PackLen); err != nil {
		return
	}
	log.Debug("packLen: %d", proto.PackLen)
	if err = binary.Read(rd, binary.BigEndian, &proto.HeaderLen); err != nil {
		return
	}
	log.Debug("headerLen: %d", proto.HeaderLen)
	if err = binary.Read(rd, binary.BigEndian, &proto.Ver); err != nil {
		return
	}
	log.Debug("ver: %d", proto.Ver)
	if err = binary.Read(rd, binary.BigEndian, &proto.Operation); err != nil {
		return
	}
	log.Debug("operation: %d", proto.Operation)
	if err = binary.Read(rd, binary.BigEndian, &proto.SeqId); err != nil {
		return
	}
	log.Debug("seqId: %d", proto.SeqId)
	if err = ReadBody(rd, proto); err != nil {
	}
	return
}

func (t *Timer) del(td *TimerData) {
	var (
		i    = td.index
		last = len(t.timers) - 1
	)
	if i < 0 || i > last || t.timers[i] != td {
		// already remove, usually by expire
		if Debug {
			log.Debug("timer del i: %d, last: %d, %p", i, last, td)
		}
		return
	}
	if i != last {
		t.swap(i, last)
		t.down(i, last)
		t.up(i)
	}
	// remove item is the last node
	t.timers[last].index = -1 // for safety
	t.timers = t.timers[:last]
	if Debug {
		log.Debug("timer: remove item key: %s, expire: %s, index: %d", td.Key, td.ExpireString(), td.index)
	}
	return
}

// Initializes a new Droplet application object
func (a *Application) Initialize() error {
	log.Debug("Initializing Droplet Application")

	log.Debug("Configuring Application")

	err := a.Configure()
	if err != nil {
		log.Error("Error configuring application: %s", err)
		panic(err)
	}

	log.Debug("Configuring Server Instance")

	err = a.ConfigureServer()
	if err != nil {
		log.Error("Error configuring server: ", err)
		panic(err)
	}

	log.Debug("Configuring Router")

	err = a.ConfigureRouter()
	if err != nil {
		log.Error("Error configuring router: ", err)
		panic(err)
	}

	a.Handler = func(w http.ResponseWriter, r *http.Request) {
		a.Router.ServeHTTP(w, r)
	}

	return nil
}

// In the case where this server is running and another one in the
// cluster stops responding, at some point this server will have to
// just write requests to disk. When the downed server comes back up,
// it's this server's responsibility to send out any writes that were
// queued up. If the yield function returns nil then the request is
// committed.
func (self *WAL) RecoverServerFromRequestNumber(requestNumber uint32, shardIds []uint32, yield func(request *protocol.Request, shardId uint32) error) error {
	// don't replay if we don't have any log files yet
	if len(self.logFiles) == 0 {
		return nil
	}

	firstIndex := 0
	firstOffset := int64(-1)
	// find the log file from which replay will start if the request
	// number is in range, otherwise replay from all log files
	if !self.isInRange(requestNumber) {
		return nil
	}

	for idx, logIndex := range self.logIndex {
		logger.Debug("Trying to find request %d in %s", requestNumber, self.logFiles[idx].file.Name())
		if firstOffset = logIndex.requestOffset(requestNumber); firstOffset != -1 {
			logger.Debug("Found reqeust %d in %s at offset %d", requestNumber, self.logFiles[idx].file.Name(), firstOffset)
			firstIndex = idx
			break
		}
	}

	// the request must be at the end of the current log file
	if firstOffset == -1 {
		firstIndex = len(self.logIndex) - 1
		firstOffset = self.logIndex[firstIndex].requestOrLastOffset(requestNumber)
	}
outer:
	for idx := firstIndex; idx < len(self.logFiles); idx++ {
		logFile := self.logFiles[idx]
		if idx > firstIndex {
			firstOffset = -1
		}
		logger.Info("Replaying from %s:%d", logFile.file.Name(), firstOffset)
		count := 0
		ch, stopChan := logFile.dupAndReplayFromOffset(shardIds, firstOffset, requestNumber)
		for {
			x := <-ch
			if x == nil {
				logger.Info("%s yielded %d requests", logFile.file.Name(), count)
				continue outer
			}

			if x.err != nil {
				return x.err
			}

			logger.Debug("Yielding request %d", x.request.GetRequestNumber())
			if err := yield(x.request, x.shardId); err != nil {
				logger.Debug("Stopping replay due to error: %s", err)
				stopChan <- struct{}{}
				return err
			}
			count++
		}
		close(stopChan)
	}
	return nil
}

func (ctx *CommunicationContext) writeLoop() {
	ctx.communicating.Add(1)
	defer ctx.communicating.Done()

	log4go.Debug("write loop started")

	for ctx.isOpen {
		tckt, ok := <-ctx.Output

		if ok {
			log4go.Debug("found new output in output channel")
			err := ctx.sender.Send(tckt.msg)

			if err != nil {
				log4go.Warn("error while sending to device: %v", err.Error())
				tckt.error <- err
			} else {
				tckt.isSend = true
				tckt.send <- tckt.msg
			}
		} else {
			log4go.Warn("output channel closed")
		}
	}

	log4go.Debug("write loop finished")
}

func (self *Shard) Query(querySpec *parser.QuerySpec, processor engine.Processor) error {
	self.closeLock.RLock()
	defer self.closeLock.RUnlock()
	if self.closed {
		return fmt.Errorf("Shard is closed")
	}
	if querySpec.IsListSeriesQuery() {
		return fmt.Errorf("List series queries should never come to the shard")
	} else if querySpec.IsDeleteFromSeriesQuery() {
		return self.executeDeleteQuery(querySpec, processor)
	}

	if !self.hasReadAccess(querySpec) {
		return errors.New("User does not have access to one or more of the series requested.")
	}

	switch t := querySpec.SelectQuery().FromClause.Type; t {
	case parser.FromClauseArray:
		log.Debug("Shard %s: running a regular query", self.db.Path())
		return self.executeArrayQuery(querySpec, processor)
	case parser.FromClauseMerge, parser.FromClauseInnerJoin:
		log.Debug("Shard %s: running a merge query", self.db.Path())
		return self.executeMergeQuery(querySpec, processor, t)
	default:
		panic(fmt.Errorf("Unknown from clause type %s", t))
	}
}

func printTimer(timer *Timer) {
	log.Debug("----------timers: %d ----------", len(timer.timers))
	for i := 0; i < len(timer.timers); i++ {
		log.Debug("timer: %s, %s, index: %d", timer.timers[i].Key, timer.timers[i].ExpireString(), timer.timers[i].index)
	}
	log.Debug("--------------------")
}

// auth for goim handshake with client, use rsa & aes.
func (server *Server) auth(rd *bufio.Reader, wr *bufio.Writer, dbm cipher.BlockMode, proto *Proto) (subKey string, heartbeat time.Duration, bucket *Bucket, channel *Channel, err error) {
	log.Debug("get auth request protocol")
	if err = server.readRequest(rd, proto); err != nil {
		return
	}
	if proto.Operation != OP_AUTH {
		log.Warn("auth operation not valid: %d", proto.Operation)
		err = ErrOperation
		return
	}
	if proto.Body, err = server.cryptor.Decrypt(dbm, proto.Body); err != nil {
		log.Error("auth decrypt client proto error(%v)", err)
		return
	}
	if subKey, heartbeat, err = server.operator.Connect(proto); err != nil {
		log.Error("operator.Connect error(%v)", err)
		return
	}
	// TODO how to reuse channel
	// register key->channel
	bucket = server.Bucket(subKey)
	channel = NewChannel(Conf.CliProto, Conf.SvrProto)
	bucket.Put(subKey, channel)
	log.Debug("send auth response protocol")
	proto.Body = nil
	proto.Operation = OP_AUTH_REPLY
	if err = server.sendResponse(wr, proto); err != nil {
		log.Error("[%s] server.SendResponse() error(%v)", subKey, err)
	}
	return
}

func (self *FoodPriceService) getDistrictFoodPrice(district string) (string, error) {
	foodPriceMsg, hitCache := districtFoodPriceMsgCache[district]
	if hitCache {
		l4g.Debug("Hit districtFoodPriceMsgCache, district: %s", district)
		return foodPriceMsg, nil
	}
	var districtFoodPrices []*DistrictFoodPrice
	entities, err := self.dbHelper.GetLatestDistrictFoodPriceEntity(district, int64(overTime))
	if err != nil {
		return "", err
	}
	if len(entities) > 0 {
		districtFoodPrices = self.convertDistrictEntityToFoodPrice(entities)
	}
	if len(districtFoodPrices) == 0 {
		var e error
		l4g.Debug("get %s food price from web", district)
		districtFoodPrices, e = FetchDistrictFoodPrice(district)
		if e != nil {
			return "", e
		}
	}
	if len(districtFoodPrices) == 0 {
		return "无记录", nil
	}
	msg := self.formatDistrictFoodPrice(districtFoodPrices, district)
	url, _ := GetDistrictFoodPriceUrl(district)
	msg = msg + fmt.Sprintf("\n详细信息请点击：%s", url)
	districtFoodPriceMsgCache[district] = msg
	return msg, nil
}

// Configure the Application
func (a *Application) Configure() error {
	log.Debug("Reading logging config file")

	log.LoadConfiguration("./loggers.xml")

	log.Debug("Reading application config file")

	file, err := ioutil.ReadFile("./config.json")
	if err != nil {
		log.Error("Error opening config file: ", err)
		panic(err)
	}

	log.Info("Config Loaded:\n\n" + string(file))

	var config Config
	err = json.Unmarshal(file, &config)
	if err != nil {
		log.Error("Error parsing config file: ", err)
		panic(err)
	}

	debug, ok := config["debug"].(bool)
	if !ok {
		debug = false
	}

	a.Config = config
	a.Debug = debug

	return nil
}

// hanleTCPConn handle a long live tcp connection.
func handleTcpConn(conn net.Conn, readerChan chan *bufio.Reader) {
	addr := conn.RemoteAddr().String()
	log.Debug("<%s> handleTcpConn routine start", addr)
	reader := newBufioReader(readerChan, conn)
	if args, err := parseCmd(reader); err == nil {
		// return buffer bufio.Reader
		putBufioReader(readerChan, reader)
		switch args[0] {
		case CmdSubscribe:
			subscribeTcpHandle(conn, args[1:])
		default:
			conn.Write(ParamErrorReply)
			log.Warn("<%s> unknown cmd \"%s\"", addr, args[0])
		}
	} else {
		// return buffer bufio.Reader
		putBufioReader(readerChan, reader)
		log.Error("<%s> parseCmd() error(%v)", addr, err)
	}
	// close the connection
	if err := conn.Close(); err != nil {
		log.Error("<%s> conn.Close() error(%v)", addr, err)
	}
	log.Debug("<%s> handleTcpConn routine stop", addr)
}

func InitWeb() {
	l4g.Debug("Initializing web routes")

	staticDir := utils.FindDir("web/static")
	l4g.Debug("Using static directory at %v", staticDir)
	api.Srv.Router.PathPrefix("/static/").Handler(http.StripPrefix("/static/",
		http.FileServer(http.Dir(staticDir))))

	api.Srv.Router.Handle("/", api.AppHandler(root)).Methods("GET")
	api.Srv.Router.Handle("/login", api.AppHandler(login)).Methods("GET")
	api.Srv.Router.Handle("/signup_team_confirm/", api.AppHandler(signupTeamConfirm)).Methods("GET")
	api.Srv.Router.Handle("/signup_team_complete/", api.AppHandler(signupTeamComplete)).Methods("GET")
	api.Srv.Router.Handle("/signup_user_complete/", api.AppHandler(signupUserComplete)).Methods("GET")

	api.Srv.Router.Handle("/logout", api.AppHandler(logout)).Methods("GET")

	api.Srv.Router.Handle("/verify", api.AppHandler(verifyEmail)).Methods("GET")
	api.Srv.Router.Handle("/find_team", api.AppHandler(findTeam)).Methods("GET")
	api.Srv.Router.Handle("/reset_password", api.AppHandler(resetPassword)).Methods("GET")

	csr := api.Srv.Router.PathPrefix("/channels").Subrouter()
	csr.Handle("/{name:[A-Za-z0-9-]+(__)?[A-Za-z0-9-]+}", api.UserRequired(getChannel)).Methods("GET")

	watchAndParseTemplates()
}

func (c *DefaultServerCodec) ReadRequestBody(rd *bufio.Reader, proto *Proto) (err error) {
	var (
		n       = int(0)
		t       = int(0)
		bodyLen = int(proto.PackLen - int32(proto.HeaderLen))
	)
	log.Debug("read body len: %d", bodyLen)
	if bodyLen > 0 {
		proto.Body = make([]byte, bodyLen)
		// no deadline, because readheader always incoming calls readbody
		for {
			if t, err = rd.Read(proto.Body[n:]); err != nil {
				log.Error("body: buf.Read() error(%v)", err)
				return
			}
			if n += t; n == bodyLen {
				log.Debug("body: rd.Read() fill ok")
				break
			} else if n < bodyLen {
				log.Debug("body: rd.Read() need %d bytes", bodyLen-n)
			} else {
				log.Error("body: readbytes %d > %d", n, bodyLen)
			}
		}
	} else {
		proto.Body = nil
	}
	return
}

func (e *Editor) RunCommand(name string, args Args) {
	// TODO?
	var (
		wnd *Window
		v   *View
	)
	if wnd = e.ActiveWindow(); wnd != nil {
		v = wnd.ActiveView()
	}

	// TODO: what's the command precedence?
	if c := e.cmdhandler.TextCommands[name]; c != nil {
		if err := e.CommandHandler().RunTextCommand(v, name, args); err != nil {
			log4go.Debug("Couldn't run textcommand: %s", err)
		}
	} else if c := e.cmdhandler.WindowCommands[name]; c != nil {
		if err := e.CommandHandler().RunWindowCommand(wnd, name, args); err != nil {
			log4go.Debug("Couldn't run windowcommand: %s", err)
		}
	} else if c := e.cmdhandler.ApplicationCommands[name]; c != nil {
		if err := e.CommandHandler().RunApplicationCommand(name, args); err != nil {
			log4go.Debug("Couldn't run applicationcommand: %s", err)
		}
	} else {
		log4go.Debug("Couldn't find command to run")
	}
}