// achieve and upload
func (b *Builder) publish(file string) (addr string, err error) {
	var path string
	if b.framework == "" {
		path, err = b.pack([]string{file}, filepath.Join(b.srcDir, ".gobuild.yml"))
	} else {
		path, err = utils.TempFile("files", "tmp-", "-"+filepath.Base(file))
		if err != nil {
			return
		}
		err = sh.Command("mv", "-v", file, path).Run()
	}
	if err != nil {
		return
	}

	// file ext<zip|tar.gz>
	suffix := ".zip"
	if strings.HasSuffix(path, ".tar.gz") {
		suffix = ".tar.gz"
	}
	go func() {
		defer func() {
			log.Debug("delete history:", b.tag)
			delete(history, b.tag)
			go func() {
				// leave 5min gap for unfinished downloading.
				time.Sleep(time.Minute * 5)
				//time.Sleep(time.Second * 5)
				os.Remove(path)
			}()
		}()
		// upload
		var cdnAddr string
		var err error
		if *environment == "development" {
			cdnAddr, err = UploadLocal(path)
		} else {
			name := fmt.Sprintf("%s-%s-%s-%s",
				filepath.Base(b.project),
				b.os, b.arch, b.ref) + suffix
			cdnAddr, err = UploadFile(path, uuid.New()+"/"+name)
		}
		if err != nil {
			return
		}
		log.Debug("upload ok:", cdnAddr)
		output := ""
		if b.wbc != nil {
			output = string(b.wbc.Bytes())
		}
		err = database.AddFile(b.pid, b.tag, cdnAddr, output)
		if err != nil {
			log.Error(err)
		}
	}()
	tmpAddr := "http://" + opts.Hostname + "/" + path
	history[b.tag] = tmpAddr
	return tmpAddr, nil
}

// Listen and forward connections
func NewTcpProxyListener(tunnel *Tunnel, port int) (listener *net.TCPListener, err error) {
	var laddr *net.TCPAddr
	if port != 0 {
		laddr, _ = net.ResolveTCPAddr("tcp", ":"+strconv.Itoa(port))
		listener, err = net.ListenTCP("tcp", laddr)
	} else {
		laddr, listener, err = freeport.ListenTCP()
	}
	if err != nil {
		return nil, err
	}
	port = laddr.Port
	//port, listener, err = listenTcpInRangePort(port, TCP_MIN_PORT, TCP_MAX_PORT)
	//if err != nil {
	//return nil, err
	//}
	// hook here
	err = hook(HOOK_TCP_POST_CONNECT, []string{
		"PORT=" + strconv.Itoa(port),
		"REMOTE_ADDR=" + tunnel.wsconn.RemoteAddr().String(),
		"CLIENT_ADDRESS=" + tunnel.wsconn.RemoteAddr().String(),
		"REMOTE_DATA=" + tunnel.data,
	})
	if err != nil {
		listener.Close()
		return
	}

	go func() {
		for {
			rconn, err := listener.AcceptTCP()
			if err != nil {
				log.Warn(err)
				break
			}
			// find proxy to where
			log.Debug("Receive new connections from", rconn.RemoteAddr())
			lconn, err := tunnel.RequestNewConn(rconn.RemoteAddr().String())
			if err != nil {
				log.Debug("request new conn err:", err)
				rconn.Close()
				continue
			}

			log.Debug("request new conn:", lconn, err)
			pc := &ProxyConn{
				lconn: lconn,
				rconn: rconn,
				stats: proxyStats,
			}
			go pc.start()
		}
	}()
	return listener, nil
}

func StartAgent(pURL *url.URL, subdomain, serverAddr string, remoteListenPort int, data string) {
	log.Debug("start proxy", pURL)
	if !regexp.MustCompile("^http[s]://").MatchString(serverAddr) {
		serverAddr = "http://" + serverAddr
	}
	sURL, err := url.Parse(serverAddr)
	if err != nil {
		log.Fatal(err)
	}

	sURL.Path = "/ws"
	log.Debug("server host:", sURL.Host)
	conn, err := net.Dial("tcp", sURL.Host)
	if err != nil {
		log.Fatal(err)
	}
	// specify remote listen port
	sURL.Scheme = "ws"
	query := sURL.Query()
	query.Add("protocol", pURL.Scheme)
	query.Add("subdomain", subdomain)
	query.Add("data", data)
	if remoteListenPort != 0 {
		query.Add("port", strconv.Itoa(remoteListenPort))
	}
	sURL.RawQuery = query.Encode()

	log.Debug(sURL)
	wsclient, _, err := websocket.NewClient(conn, sURL, nil, 1024, 1024)
	if err != nil {
		log.Fatal(err)
	}
	defer wsclient.Close()
	go idleWsSend(wsclient)

	for {
		var msg Msg
		if err := wsclient.ReadJSON(&msg); err != nil {
			fmt.Println("client exit: " + err.Error())
			break
		}
		log.Debug("recv:", msg)

		// sURL: serverURL
		rnl := NewRevNetListener()
		go handleRevConn(pURL, rnl)
		handleWsMsg(msg, sURL, rnl)
	}
}

func (p *ProxyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// http://stackoverflow.com/questions/6899069/why-are-request-url-host-and-scheme-blank-in-the-development-server
	r.URL.Scheme = "http" // ??
	r.URL.Host = r.Host   // ??
	log.Debug("URL path:", r.URL.Path)
	log.Debugf("proxy lists: %v", p.revProxies)
	if rpx, ok := p.revProxies[r.Host]; ok {
		log.Debug("server http rev proxy")
		rpx.ServeHTTP(w, r)
		return
	}

	h, _ := p.Handler(r)
	h.ServeHTTP(w, r)
}

func (this *DirCache) Cache(cacheRootPath string, cacheResultFile string) (fileCount int) {
	if _, err := os.Stat(cacheResultFile); err != nil {
		log.Info(fmt.Sprintf("No cache file `%s' found, will create one", cacheResultFile))
	} else {
		os.Remove(cacheResultFile + ".old")
		if rErr := os.Rename(cacheResultFile, cacheResultFile+".old"); rErr != nil {
			log.Error(fmt.Sprintf("Unable to rename cache file, plz manually delete `%s' and `%s.old'",
				cacheResultFile, cacheResultFile))
			log.Error(rErr)
			return
		}
	}
	cacheResultFileH, err := os.OpenFile(cacheResultFile, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0666)
	if err != nil {
		log.Error(fmt.Sprintf("Failed to open cache file `%s'", cacheResultFile))
		return
	}
	defer cacheResultFileH.Close()
	bWriter := bufio.NewWriter(cacheResultFileH)
	walkStart := time.Now()
	log.Debug(fmt.Sprintf("Walk `%s' start from `%s'", cacheRootPath, walkStart.String()))
	filepath.Walk(cacheRootPath, func(path string, fi os.FileInfo, err error) error {
		var retErr error
		//log.Debug(fmt.Sprintf("Walking through `%s'", cacheRootPath))
		if !fi.IsDir() {
			relPath := strings.TrimPrefix(strings.TrimPrefix(path, cacheRootPath), string(os.PathSeparator))
			fsize := fi.Size()
			//Unit is 100ns
			flmd := fi.ModTime().UnixNano() / 100
			//log.Debug(fmt.Sprintf("Hit file `%s' size: `%d' mode time: `%d`", relPath, fsize, flmd))
			fmeta := fmt.Sprintln(fmt.Sprintf("%s\t%d\t%d", relPath, fsize, flmd))
			if _, err := bWriter.WriteString(fmeta); err != nil {
				log.Error(fmt.Sprintf("Failed to write data `%s' to cache file", fmeta))
				retErr = err
			}
			fileCount += 1
		}
		return retErr
	})
	if err := bWriter.Flush(); err != nil {
		log.Error(fmt.Sprintf("Failed to flush to cache file `%s'", cacheResultFile))
	}

	walkEnd := time.Now()
	log.Debug(fmt.Sprintf("Walk `%s' end at `%s'", cacheRootPath, walkEnd.String()))
	log.Debug(fmt.Sprintf("Walk `%s' last for `%s'", cacheRootPath, time.Since(walkStart)))
	return
}

func Register(dict *Dict, nsprefix string, self interface{}) (ctx RegisterCtx) {

	typ := reflect.TypeOf(self)
	selfv := reflect.ValueOf(self)

	nmethod := typ.NumMethod()

	for i := 0; i < nmethod; i++ {
		method := typ.Method(i)
		mtype := method.Type
		mname := method.Name
		if mtype.PkgPath() != "" || !strings.HasPrefix(mname, "Py_") {
			continue
		}
		nin := mtype.NumIn()
		name := mname[3:]
		fullname := nsprefix + name
		if nin == 3 && sigMatches(mtype, typTernaryCallFunc) || nin == 2 && sigMatches(mtype, typBinaryCallFunc) {
			closure := &Closure{selfv, method.Func}
			f := closure.NewFunction(fullname, nin, "")
			dict.SetItemString(name, f)
			f.Decref()
			ctx = append(ctx, closure)
			log.Debug("Register", fullname)
		} else {
			log.Warnf("Invalid signature of method %s, register failed", fullname)
			continue
		}
	}
	return
}

func (p *ProxyConn) pipe(src, dst net.Conn) chan error {
	//data direction
	errch := make(chan error, 1)
	islocal := src == p.lconn

	//directional copy (64k buffer)
	buff := make([]byte, 0xffff)
	go func() {
		for {
			n, err := src.Read(buff)
			if err != nil {
				errch <- err
				return
			}
			b := buff[:n]

			//write out result
			n, err = dst.Write(b)
			if err != nil {
				errch <- err
				log.Printf("Write failed '%s'\n", err)
				return
			}
			log.Debug("pipe --> local:", islocal, "write:", n) //, string(b[:n]))
			if islocal {
				p.sentBytes += uint64(n)
				p.stats.sentBytes += uint64(n)
			} else {
				p.receivedBytes += uint64(n)
				p.stats.receivedBytes += uint64(n)
			}
		}
	}()
	return errch
}

// output websocket
func WsBuildServer(ws *websocket.Conn) {
	defer ws.Close()
	var err error
	recvMsg := new(RecvMsg)
	sendMsg := new(SendMsg)
	err = websocket.JSON.Receive(ws, &recvMsg)
	if err != nil {
		sendMsg.Error = err
		websocket.JSON.Send(ws, sendMsg)
		utils.Debugf("read json error: %v", err)
		return
	}

	name := ws.RemoteAddr().String()
	log.Debug(name)

	sout := NewStreamOutput(recvMsg.Project, recvMsg.Branch, recvMsg.GOOS, recvMsg.GOARCH)
	defer sout.Close()

	sendMsg.Data = sout.BufferStr
	err = websocket.JSON.Send(ws, sendMsg)
	if err != nil {
		utils.Debugf("send first sendMsg error: %v", err)
		return
	}

	// send the rest outputs
	buf := make([]byte, 100)
	for {
		n, err := sout.Reader.Read(buf)
		if n > 0 {
			sendMsg.Data = string(buf[:n])
			deadline := time.Now().Add(time.Second * 1)
			ws.SetWriteDeadline(deadline)
			if er := websocket.JSON.Send(ws, sendMsg); er != nil {
				log.Debug("write failed timeout, user logout")
				return
			}
		}
		if err != nil {
			return
		}
	}
}

func (this *ListBucket) List(bucket string, prefix string, listResultFile string) (retErr error) {
	fp, openErr := os.OpenFile(listResultFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
	if openErr != nil {
		retErr = openErr
		log.Error(fmt.Sprintf("Failed to open list result file `%s'", listResultFile))
		return
	}
	defer fp.Close()
	bw := bufio.NewWriter(fp)

	mac := digest.Mac{this.AccessKey, []byte(this.SecretKey)}
	client := rsf.New(&mac)
	marker := ""
	limit := 1000
	run := true
	maxRetryTimes := 5
	retryTimes := 1
	for run {
		entries, markerOut, err := client.ListPrefix(nil, bucket, prefix, marker, limit)
		if err != nil {
			if err == io.EOF {
				run = false
			} else {
				log.Error(fmt.Sprintf("List error for marker `%s'", marker), err)
				if retryTimes <= maxRetryTimes {
					log.Debug(fmt.Sprintf("Retry list for marker `%s' for `%d' time", marker, retryTimes))
					retryTimes += 1
					continue
				} else {
					log.Error(fmt.Sprintf("List failed too many times for `%s'", marker))
					break
				}
			}
		} else {
			retryTimes = 1
			if markerOut == "" {
				run = false
			} else {
				marker = markerOut
			}
		}
		//append entries
		for _, entry := range entries {
			lineData := fmt.Sprintf("%s\t%d\t%s\t%d\t%s\t%s\r\n", entry.Key, entry.Fsize, entry.Hash, entry.PutTime, entry.MimeType, entry.EndUser)
			_, wErr := bw.WriteString(lineData)
			if wErr != nil {
				log.Error(fmt.Sprintf("Write line data `%s' to list result file failed.", lineData))
			}
		}
		fErr := bw.Flush()
		if fErr != nil {
			log.Error("Flush data to list result file error", err)
		}
	}
	return
}

func closeWrite(c net.Conn) error {
	if x, ok := c.(interface {
		CloseWrite() error
	}); ok {
		return x.CloseWrite()
	} else {
		log.Debug("force close", c)
		return c.Close()
	}
}

func (this *AliListBucket) ListBucket(listResultFile string) (err error) {
	//open result file
	mode := os.O_CREATE | os.O_TRUNC | os.O_WRONLY
	fp, openErr := os.OpenFile(listResultFile, mode, 0666)
	if openErr != nil {
		err = openErr
		return
	}
	defer fp.Close()
	bw := bufio.NewWriter(fp)
	//list bucket by prefix
	marker := ""
	prefixLen := len(this.Prefix)
	ossClient := oss.NewClient(this.DataCenter, this.AccessKeyId, this.AccessKeySecret, 0)
	maxRetryTimes := 5
	retryTimes := 1
	for {
		lbr, lbrErr := ossClient.GetBucket(this.Bucket, this.Prefix, marker, "", "")
		if lbrErr != nil {
			err = lbrErr
			log.Error("Parse list result error, ", "marker=[", marker, "]", lbrErr)
			if retryTimes <= maxRetryTimes {
				log.Debug("Retry marker=", marker, "] for ", retryTimes, "time...")
				retryTimes += 1
				continue
			} else {
				break
			}
		} else {
			retryTimes = 1
		}
		for _, object := range lbr.Contents {
			lmdTime, lmdPErr := time.Parse("2006-01-02T15:04:05.999Z", object.LastModified)
			if lmdPErr != nil {
				log.Error("Parse object last modified error, ", lmdPErr)
				lmdTime = time.Now()
			}
			bw.WriteString(fmt.Sprintln(fmt.Sprintf("%s\t%d\t%d", object.Key[prefixLen:], object.Size, lmdTime.UnixNano()/100)))
		}
		if !lbr.IsTruncated {
			break
		}
		marker = lbr.NextMarker
	}
	fErr := bw.Flush()
	if fErr != nil {
		log.Error("Write data to buffer writer failed", fErr)
		err = fErr
		return
	}
	return err
}

func (p *proxyConn) start() {
	defer p.lconn.Close()
	defer p.rconn.Close()

	// FIXME: may need to set a flag
	if tcpconn, ok := p.lconn.(*net.TCPConn); ok {
		tcpconn.SetNoDelay(true)
	}
	if tcpconn, ok := p.rconn.(*net.TCPConn); ok {
		tcpconn.SetNoDelay(true)
	}
	// p.lconn.SetNoDelay(true)
	// p.rconn.SetNoDelay(true)

	//display both ends
	// log.Printf("Opened %s >>> %s", p.lconn.RemoteAddr().String(), p.rconn.RemoteAddr().String())
	//bidirectional copy
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		ch1 := p.pipe(p.lconn, p.rconn)
		<-ch1
		closeRead(p.lconn)
		closeWrite(p.rconn)
		log.Debug("close local -> remote")
		wg.Done()
	}()
	go func() {
		ch2 := p.pipe(p.rconn, p.lconn)
		<-ch2
		closeRead(p.rconn)
		closeWrite(p.lconn)
		log.Debug("close remote -> local")
		wg.Done()
	}()
	wg.Wait()
	//wait for close...
	// log.Printf("Closed (%d bytes sent, %d bytes recieved)", p.sentBytes, p.receivedBytes)
}

func (b *Builder) pack(bins []string, rcfile string) (path string, err error) {
	log.Debug(bins)
	log.Debug(rcfile)
	data, err := ioutil.ReadFile(rcfile)
	if err != nil {
		log.Debug("use default rc")
		data, err = ioutil.ReadFile("public/gobuildrc")
		if err != nil {
			log.Error(err)
		}
	}
	ass := new(Assembly)
	err = goyaml.Unmarshal(data, ass)
	if err != nil {
		return
	}
	dir := filepath.Dir(rcfile)
	fs, err := ioutil.ReadDir(dir)
	if err != nil {
		return
	}
	var includes = bins // this may change slice bins
	for _, f := range fs {
		var ok = false
		for _, patten := range ass.Includes {
			if match(patten, f.Name()) {
				ok = true
				break
			}
		}

		if ok {
			includes = append(includes, filepath.Join(dir, f.Name()))
		}
	}
	return pkgZip(dir, includes)
}

func runShellScript(ctx *macaron.Context, script string) ([]byte, error) {
	ctx.Req.ParseForm()
	//log.Println(ctx.Req.Form)
	envs := map[string]string{
		"REQ_PATH":   ctx.Req.URL.Path,
		"REQ_URI":    ctx.Req.RequestURI,
		"REQ_METHOD": ctx.Req.Method,
	}
	for key, vals := range ctx.Req.Form {
		log.Debug("Form value:", key, vals)
		envs["FORM_"+key] = vals[0]
	}
	for key, vals := range ctx.Req.PostForm {
		log.Debug("Form value:", key, vals)
		envs["POST_FORM_"+key] = vals[0]
	}
	environ := os.Environ()
	for key, val := range envs {
		environ = append(environ, key+"="+val)
	}
	cmd := exec.Command("/bin/bash", script) //.Output()
	cmd.Env = environ
	return cmd.Output()
}

func batchChgm(client rs.Client, entries []qshell.ChgmEntryPath) {
	ret, err := qshell.BatchChgm(client, entries)
	if err != nil {
		log.Error("Batch chgm error", err)
	}
	if len(ret) > 0 {
		for i, entry := range entries {
			item := ret[i]
			if item.Data.Error != "" {
				log.Error(fmt.Sprintf("Chgm '%s' => '%s' Failed, Code :%d", entry.Key, entry.MimeType, item.Code))
			} else {
				log.Debug(fmt.Sprintf("Chgm '%s' => '%s' Success, Code :%d", entry.Key, entry.MimeType, item.Code))
			}
		}
	}
}

func batchDelete(client rs.Client, entries []rs.EntryPath) {
	ret, err := qshell.BatchDelete(client, entries)
	if err != nil {
		log.Error("Batch delete error", err)
	}
	if len(ret) > 0 {
		for i, entry := range entries {
			item := ret[i]
			if item.Data.Error != "" {
				log.Error(fmt.Sprintf("Delete '%s' => '%s' Failed, Code: %d", entry.Bucket, entry.Key, item.Code))
			} else {
				log.Debug(fmt.Sprintf("Delete '%s' => '%s' Success, Code: %d", entry.Bucket, entry.Key, item.Code))
			}
		}
	}
}

func batchRename(client rs.Client, entries []qshell.RenameEntryPath) {
	ret, err := qshell.BatchRename(client, entries)
	if err != nil {
		log.Error("Batch rename error", err)
	}
	if len(ret) > 0 {
		for i, entry := range entries {
			item := ret[i]
			if item.Data.Error != "" {
				log.Error(fmt.Sprintf("Rename '%s' => '%s' Failed, Code :%d", entry.OldKey, entry.NewKey, item.Code))
			} else {
				log.Debug(fmt.Sprintf("Rename '%s' => '%s' Success, Code :%d", entry.OldKey, entry.NewKey, item.Code))
			}
		}
	}
}

// download src
func (b *Builder) get() (err error) {
	log.Debug("start get src to:", b.srcDir)
	exists := beeutils.FileExists(b.srcDir)
	b.sh.Command("go", "version").Run()
	if !exists {
		err = b.sh.Command("go", "get", "-v", "-d", b.project).Run()
		if err != nil {
			return
		}
	}
	b.sh.SetDir(b.srcDir)
	if b.ref == "-" {
		b.ref = "master"
	}
	// get code from remote
	if err = b.sh.Command("git", "fetch", "origin").Run(); err != nil {
		return
	}
	// change branch
	if err = b.sh.Command("git", "checkout", "-q", b.ref).Run(); err != nil {
		return
	}
	// update code
	if err = b.sh.Command("git", "merge", "origin/"+b.ref).Run(); err != nil {
		log.Warn("git merge error:", err)
		//return
	}
	// get sha
	out, err := sh.Command("git", "rev-parse", "HEAD", sh.Dir(b.srcDir)).Output()
	if err != nil {
		return
	}
	b.sha = strings.TrimSpace(string(out))

	// parse .gobuild
	b.rc = new(Assembly)
	rcfile := "public/gobuildrc"
	if b.sh.Test("f", ".gobuild") {
		rcfile = filepath.Join(b.srcDir, ".gobuild")
	}
	data, err := ioutil.ReadFile(rcfile)
	if err != nil {
		return
	}
	err = goyaml.Unmarshal(data, b.rc)
	return
}

func pkgZip(root string, files []string) (path string, err error) {
	log.Info("package to zip:", path)
	tmpFile, err := utils.TempFile("files", "tmp-", "-"+filepath.Base(root)+".zip")
	if err != nil {
		return
	}

	z, err := zip.Create(tmpFile)
	if err != nil {
		return
	}
	for _, f := range files {
		var save string
		if f == "" {
			continue
		}
		// binary file use abspath
		//fmt.Println(root, f)
		if strings.HasSuffix(f, root) {
			save = f[len(root):]
		} else {
			save = filepath.Base(f)
		}
		info, er := os.Stat(f)
		if er != nil {
			continue
		}
		log.Debug("add", save, f)
		if info.IsDir() {
			if err = z.AddDir(save, f); err != nil {
				return
			}
		} else {
			if err = z.AddFile(save, f); err != nil {
				return
			}
		}
	}
	if err = z.Close(); err != nil {
		log.Error(err)
		return
	}
	return tmpFile, nil

}

func deployPackage(pkgName, path string, binDir string) error {
	cmd := exec.Command("unzip", "-o", "-d", getInsPath("opt", pkgName), path)
	log.Debug("zip command:", cmd.Args)
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	if err := cmd.Run(); err != nil {
		return err
	}
	baseName := filepath.Base(pkgName)
	symlink := filepath.Join(GOBIN, baseName)

	prompt("Symlink %v", symlink)
	// for linux and darwin
	os.Remove(symlink)

	// TODO: need to resolve multi binaries
	return os.Symlink(getInsPath("opt", pkgName, filepath.Base(pkgName)), symlink)
}