func (node *Node) httpGCData(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	count, err := node.doGC(ctx)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		if count > 0 {
			fmt.Fprintf(w, "Partial GC: %d objects deleted\n", count)
		}
		return
	}
	fmt.Fprintln(w, count)
}
POST /data/gc
garbage collect orphan data objects that are not referenced by any statement
package main
import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	mux "github.com/gorilla/mux"
	p2p_crypto "github.com/libp2p/go-libp2p-crypto"
	p2p_peer "github.com/libp2p/go-libp2p-peer"
	p2p_pstore "github.com/libp2p/go-libp2p-peerstore"
	mc "github.com/mediachain/concat/mc"
	mcq "github.com/mediachain/concat/mc/query"
	pb "github.com/mediachain/concat/proto"
	multihash "github.com/multiformats/go-multihash"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"regexp"
	"strconv"
	"strings"
	"time"
)
func apiError(w http.ResponseWriter, status int, err error) {
	w.WriteHeader(status)
	fmt.Fprintf(w, "Error: %s\n", err.Error())
}
func apiNetError(w http.ResponseWriter, err error) {
	switch err {
	case UnknownPeer:
		apiError(w, http.StatusNotFound, err)
	default:
		apiError(w, http.StatusInternalServerError, err)
	}
}
func (node *Node) httpId(w http.ResponseWriter, r *http.Request) {
	ninfo := NodeInfo{node.PeerIdentity.Pretty(), node.publisher.Pretty(), node.info}
	err := json.NewEncoder(w).Encode(ninfo)
	if err != nil {
		log.Printf("Error writing response body: %s", err.Error())
	}
}
func (node *Node) httpRemoteId(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	ninfo, err := node.doRemoteId(ctx, pid)
	if err != nil {
		apiNetError(w, err)
		return
	}
	err = json.NewEncoder(w).Encode(ninfo)
	if err != nil {
		log.Printf("Error writing response body: %s", err.Error())
	}
}
func (node *Node) httpNetAddr(w http.ResponseWriter, r *http.Request) {
	addrs := node.netAddrs()
	for _, addr := range addrs {
		fmt.Fprintln(w, addr.String())
	}
}
func (node *Node) httpNetPeerAddr(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	addrs := node.netPeerAddrs(pid)
	for _, addr := range addrs {
		fmt.Fprintln(w, addr.String())
	}
}
func (node *Node) httpNetConns(w http.ResponseWriter, r *http.Request) {
	peers := node.netConns()
	for _, peer := range peers {
		fmt.Fprintln(w, mc.FormatHandle(peer))
	}
}
func (node *Node) httpNetLookup(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	pinfo, err := node.doLookup(r.Context(), pid)
	if err != nil {
		apiNetError(w, err)
		return
	}
	for _, addr := range pinfo.Addrs {
		fmt.Fprintln(w, addr.String())
	}
}
func (node *Node) httpNetIdentify(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	nid, err := node.netIdentify(ctx, pid)
	if err != nil {
		apiNetError(w, err)
		return
	}
	err = json.NewEncoder(w).Encode(nid)
	if err != nil {
		log.Printf("Error writing response body: %s", err.Error())
	}
}
func (node *Node) httpNetPing(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	dt, err := node.netPing(ctx, pid)
	if err != nil {
		apiNetError(w, err)
		return
	}
	fmt.Fprintln(w, dt)
}
func (node *Node) httpNetFindPeers(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	ch, err := node.netFindPeers(ctx)
	if err != nil {
		apiNetError(w, err)
		return
	}
	for pinfo := range ch {
		fmt.Fprintln(w, pinfo.ID.Pretty())
	}
}
func (node *Node) httpPing(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	err = node.doPing(ctx, pid)
	if err != nil {
		apiNetError(w, err)
		return
	}
	fmt.Fprintln(w, "OK")
}
func (node *Node) httpDirList(w http.ResponseWriter, r *http.Request) {
	node.httpDirListImpl(w, r, false)
}
func (node *Node) httpDirListAll(w http.ResponseWriter, r *http.Request) {
	node.httpDirListImpl(w, r, true)
}
func (node *Node) httpDirListImpl(w http.ResponseWriter, r *http.Request, inclSelf bool) {
	vars := mux.Vars(r)
	ns := vars["namespace"]
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	peers, err := node.doDirList(ctx, ns)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	mypid := node.PeerIdentity.Pretty()
	for _, peer := range peers {
		if peer != mypid || inclSelf {
			fmt.Fprintln(w, peer)
		}
	}
}
func (node *Node) httpDirListNS(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	nss, err := node.doDirListNS(ctx)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	for _, ns := range nss {
		fmt.Fprintln(w, ns)
	}
}
func (node *Node) httpDirListMF(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	entity := vars["entity"]
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	mfs, err := node.doDirListMF(ctx, entity)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	enc := json.NewEncoder(w)
	for _, mf := range mfs {
		err = enc.Encode(mf)
		if err != nil {
			log.Printf("Error encoding manifest result: %s", err.Error())
			return
		}
	}
}
var nsrx *regexp.Regexp
func init() {
	rx, err := regexp.Compile("^[a-zA-Z0-9-]+([.][a-zA-Z0-9-]+)*$")
	if err != nil {
		log.Fatal(err)
	}
	nsrx = rx
}
func (node *Node) httpPublish(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	ns := vars["namespace"]
	if !nsrx.Match([]byte(ns)) {
		apiError(w, http.StatusBadRequest, BadNamespace)
		return
	}
	dec := json.NewDecoder(r.Body)
	stmts := make([]interface{}, 0, 1024)
loop:
	for {
		sbody := new(pb.SimpleStatement)
		err := dec.Decode(sbody)
		switch {
		case err == io.EOF:
			break loop
		case err != nil:
			apiError(w, http.StatusBadRequest, err)
			return
		default:
			stmts = append(stmts, sbody)
		}
	}
	if len(stmts) == 0 {
		return
	}
	sids, err := node.doPublishBatch(ns, stmts)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	for _, sid := range sids {
		fmt.Fprintln(w, sid)
	}
}
func (node *Node) httpPublishCompound(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	ns := vars["namespace"]
	if !nsrx.Match([]byte(ns)) {
		apiError(w, http.StatusBadRequest, BadNamespace)
		return
	}
	comb := vars["combine"]
	clen, err := strconv.Atoi(comb)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	dec := json.NewDecoder(r.Body)
	stmts := make([]interface{}, 0, 1000/clen)
	body := make([]*pb.SimpleStatement, 0, clen)
loop:
	for {
		for x := 0; x < clen; x++ {
			sbody := new(pb.SimpleStatement)
			err = dec.Decode(sbody)
			switch {
			case err == io.EOF:
				break loop
			case err != nil:
				apiError(w, http.StatusBadRequest, err)
				return
			default:
				body = append(body, sbody)
			}
		}
		stmt := &pb.CompoundStatement{body}
		stmts = append(stmts, stmt)
		body = make([]*pb.SimpleStatement, 0, clen)
	}
	if len(body) > 0 {
		stmt := &pb.CompoundStatement{body}
		stmts = append(stmts, stmt)
	}
	if len(stmts) == 0 {
		return
	}
	sids, err := node.doPublishBatch(ns, stmts)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	for _, sid := range sids {
		fmt.Fprintln(w, sid)
	}
}
func (node *Node) httpImport(w http.ResponseWriter, r *http.Request) {
	const batch = 1024
	var err error
	count := 0
	dec := json.NewDecoder(r.Body)
	pkcache := make(map[string]p2p_crypto.PubKey)
	stmts := make([]*pb.Statement, 0, batch)
loop:
	for {
		stmt := new(pb.Statement)
		err = dec.Decode(stmt)
		switch {
		case err == io.EOF:
			break loop
		case err != nil:
			goto import_error
		}
		stmts = append(stmts, stmt)
		if len(stmts) >= batch {
			var xcount int
			xcount, err = node.doImport(stmts, pkcache)
			count += xcount
			if err != nil {
				goto import_error
			}
			stmts = stmts[:0]
		}
	}
	if len(stmts) > 0 {
		var xcount int
		xcount, err = node.doImport(stmts, pkcache)
		count += xcount
		if err != nil {
			goto import_error
		}
	}
	fmt.Fprintln(w, count)
	return
import_error:
	apiError(w, http.StatusBadRequest, err)
	if count > 0 {
		fmt.Fprintf(w, "Partial import: %d statements merged\n", count)
	}
}
func (node *Node) httpStatement(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	id := vars["statementId"]
	stmt, err := node.db.Get(id)
	if err != nil {
		switch err {
		case UnknownStatement:
			apiError(w, http.StatusNotFound, err)
			return
		default:
			apiError(w, http.StatusInternalServerError, err)
			return
		}
	}
	err = json.NewEncoder(w).Encode(stmt)
	if err != nil {
		log.Printf("Error writing response body: %s", err.Error())
	}
}
func (node *Node) httpQuery(w http.ResponseWriter, r *http.Request) {
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("http/query: Error reading request body: %s", err.Error())
		return
	}
	q, err := mcq.ParseQuery(string(body))
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	if q.Op != mcq.OpSelect {
		apiError(w, http.StatusBadRequest, BadQuery)
		return
	}
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	ch, err := node.db.QueryStream(ctx, q)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	enc := json.NewEncoder(w)
	for obj := range ch {
		err = enc.Encode(obj)
		if err != nil {
			log.Printf("Error encoding query result: %s", err.Error())
			return
		}
	}
}
func (node *Node) httpRemoteQuery(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("http/query: Error reading request body: %s", err.Error())
		return
	}
	q := string(body)
	qq, err := mcq.ParseQuery(q)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	if qq.Op != mcq.OpSelect {
		apiError(w, http.StatusBadRequest, BadQuery)
		return
	}
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	ch, err := node.doRemoteQuery(ctx, pid, q)
	if err != nil {
		apiNetError(w, err)
		return
	}
	enc := json.NewEncoder(w)
	for obj := range ch {
		err = enc.Encode(obj)
		if err != nil {
			log.Printf("Error encoding query result: %s", err.Error())
			return
		}
	}
}
func (node *Node) httpMerge(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("http/merge: Error reading request body: %s", err.Error())
		return
	}
	q := string(body)
	qq, err := mcq.ParseQuery(q)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	if !qq.IsSimpleSelect("*") {
		apiError(w, http.StatusBadRequest, BadQuery)
		return
	}
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	count, ocount, err := node.doMerge(ctx, pid, q)
	if err != nil {
		apiNetError(w, err)
		if count > 0 {
			fmt.Fprintf(w, "Partial merge: %d statements merged\n", count)
		}
		if ocount > 0 {
			fmt.Fprintf(w, "Partial merge: %d objects merged\n", ocount)
		}
		return
	}
	fmt.Fprintln(w, count)
	fmt.Fprintln(w, ocount)
}
func (node *Node) httpPush(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("http/push: Error reading request body: %s", err.Error())
		return
	}
	q := string(body)
	qq, err := mcq.ParseQuery(q)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	if !qq.IsSimpleSelect("*") {
		apiError(w, http.StatusBadRequest, BadQuery)
		return
	}
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	count, ocount, err := node.doPush(ctx, pid, qq)
	if err != nil {
		apiNetError(w, err)
		if count > 0 {
			fmt.Fprintf(w, "Partial push: %d statements merged\n", count)
		}
		if ocount > 0 {
			fmt.Fprintf(w, "Partial push: %d objects merged\n", ocount)
		}
		if count < 0 {
			fmt.Fprintf(w, "Incomplete push: some statements may have been merged\n")
		}
		return
	}
	fmt.Fprintln(w, count)
	fmt.Fprintln(w, ocount)
}
func (node *Node) httpDelete(w http.ResponseWriter, r *http.Request) {
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("http/query: Error reading request body: %s", err.Error())
		return
	}
	q, err := mcq.ParseQuery(string(body))
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	if q.Op != mcq.OpDelete {
		apiError(w, http.StatusBadRequest, BadQuery)
		return
	}
	count, err := node.db.Delete(q)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		if count > 0 {
			fmt.Fprintf(w, "Partial delete: %d statements deleted\n", count)
		}
		return
	}
	fmt.Fprintln(w, count)
}
func (node *Node) httpVacuumIncremental(w http.ResponseWriter, r *http.Request) {
	err := node.db.Vacuum(false)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	fmt.Fprintln(w, "OK")
}
func (node *Node) httpVacuumFull(w http.ResponseWriter, r *http.Request) {
	err := node.db.Vacuum(true)
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	fmt.Fprintln(w, "OK")
}
type DataObject struct {
	Data []byte `json:"data"`
}
func (node *Node) httpGetData(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	key58 := vars["objectId"]
	key, err := multihash.FromB58String(key58)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	data, err := node.ds.Get(Key(key))
	if err != nil {
		apiError(w, http.StatusInternalServerError, err)
		return
	}
	if data == nil {
		apiError(w, http.StatusNotFound, UnknownObject)
		return
	}
	dao := DataObject{data}
	err = json.NewEncoder(w).Encode(dao)
	if err != nil {
		log.Printf("Error writing response body: %s", err.Error())
	}
}
func (node *Node) httpGetDataBatch(w http.ResponseWriter, r *http.Request) {
	enc := json.NewEncoder(w)
	scanner := bufio.NewScanner(r.Body)
	for scanner.Scan() {
		key58 := scanner.Text()
		key, err := multihash.FromB58String(key58)
		if err != nil {
			enc.Encode(StreamError{err.Error()})
			return
		}
		data, err := node.ds.Get(Key(key))
		if err != nil {
			enc.Encode(StreamError{err.Error()})
			return
		}
		err = enc.Encode(DataObject{data})
		if err != nil {
			log.Printf("Error writing response body: %s", err.Error())
			return
		}
	}
}
func (node *Node) httpPutData(w http.ResponseWriter, r *http.Request) {
	dec := json.NewDecoder(r.Body)
	batch := make([][]byte, 0, 1024)
	var dao DataObject
loop:
	for {
		err := dec.Decode(&dao)
		switch {
		case err == io.EOF:
			break loop
		case err != nil:
			apiError(w, http.StatusBadRequest, err)
			return
		default:
			batch = append(batch, dao.Data)
			dao.Data = nil
		}
	}
	keys := make([]Key, len(batch))
	for x, data := range batch {
		key, err := node.ds.Put(data)
		if err != nil {
			apiError(w, http.StatusInternalServerError, err)
			return
		}
		keys[x] = key
	}
	for _, key := range keys {
		fmt.Fprintln(w, multihash.Multihash(key).B58String())
	}
}
func (node *Node) httpMergeData(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	peerId := vars["peerId"]
	pid, err := p2p_peer.IDB58Decode(peerId)
	if err != nil {
		apiError(w, http.StatusBadRequest, err)
		return
	}
	keys := make(map[string]Key)
	scanner := bufio.NewScanner(r.Body)
	for scanner.Scan() {
		err = node.mergeObjectKey(scanner.Text(), keys)
		if err != nil {
			apiError(w, http.StatusBadRequest, err)
			return
		}
	}
	count, err := node.doRawMerge(r.Context(), pid, keys)
	if err != nil {
		apiNetError(w, err)
		if count > 0 {
			fmt.Fprintf(w, "Partial merge: %d objects merged\n", count)
		}
		return
	}
	fmt.Fprintln(w, count)
}