Adds a Blob server for getting teh blobs

bot.go

@@ -8,11 +8,16 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net"
+	"net/http"
 	"os"
 	"sync"
 	"time"
 
 	"github.com/cryptix/go/logging"
+	"github.com/go-kit/kit/log/level"
+	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
 	"go.cryptoscope.co/luigi"
 	"go.cryptoscope.co/margaret"
@@ -21,6 +26,7 @@ import (
 	"go.cryptoscope.co/secretstream"
 
 	"go.cryptoscope.co/ssb"
+	"go.cryptoscope.co/ssb/blobstore"
 	"go.cryptoscope.co/ssb/multilogs"
 	"go.cryptoscope.co/ssb/repo"
 	mksbot "go.cryptoscope.co/ssb/sbot"
@@ -210,30 +216,29 @@ func BlobsGet(refStr string) ([]byte, error) {
 
 	buf := new(bytes.Buffer)
 	_, err = buf.ReadFrom(r)
-
 	if err != nil {
 		return nil, err
 	}
 	var slice = buf.Bytes()
 	return slice, nil
 }
 
-func Peers() ([]byte, error){
+func Peers() ([]byte, error) {
 	w := &bytes.Buffer{}
-  status, err := theBot.Status()
-  if err != nil {
-    return nil, err
-  }
-  var peers = status.Peers
-
-  // Apparently peers can be null and that's painful for converting to json and back
-  // If peers is null then set it to an empty list.
-  if peers == nil {
-    peers = []ssb.PeerStatus{}
-  }
-  if err := json.NewEncoder(w).Encode(peers); err != nil {
-    return nil, err
-  }
+	status, err := theBot.Status()
+	if err != nil {
+		return nil, err
+	}
+	var peers = status.Peers
+
+	// Apparently peers can be null and that's painful for converting to json and back
+	// If peers is null then set it to an empty list.
+	if peers == nil {
+		peers = []ssb.PeerStatus{}
+	}
+	if err := json.NewEncoder(w).Encode(peers); err != nil {
+		return nil, err
+	}
 
 	return w.Bytes(), nil
 }
@@ -275,6 +280,83 @@ func Start(repoPath string) {
 	checkFatal(err)
 	log.Log("event", "serving", "ID", id.Ref(), "addr", listenAddr)
 
+	// open listener first (so we can error out if the port is taken)
+	// TODO: could maybe be localhost?
+	lis, err := net.Listen("tcp", "0.0.0.0:8091")
+	checkFatal(errors.Wrap(err, "blobsServ listen failed"))
+
+	go func() {
+		r := mux.NewRouter()
+
+		r.HandleFunc("/blobs/{blobHash}", func(w http.ResponseWriter, r *http.Request) {
+			vars := mux.Vars(r)
+
+			blobHash := vars["blobHash"]
+			blobRef, err := ssb.ParseBlobRef(blobHash)
+			if err != nil {
+				http.Error(w, "bad request", http.StatusBadRequest)
+				level.Error(log).Log("msg", "failed to parse url parameter", "err", err)
+				return
+			}
+
+			blobReader, err := theBot.BlobStore.Get(blobRef)
+			if err != nil {
+				if errors.Cause(err) == blobstore.ErrNoSuchBlob {
+					theBot.WantManager.Want(blobRef)
+
+					received := make(chan struct{})
+					cancel := theBot.BlobStore.Changes().Register(luigi.FuncSink(func(ctx context.Context, v interface{}, err error) error {
+						if err != nil {
+							if luigi.IsEOS(err) {
+								return nil
+							}
+							return err
+						}
+
+						n, ok := v.(ssb.BlobStoreNotification)
+						if !ok {
+							return errors.Errorf("blob change: unhandled notification type: %T", v)
+						}
+
+						if n.Op != ssb.BlobStoreOpPut {
+							return nil
+						}
+
+						if !n.Ref.Equal(blobRef) {
+							return nil // yet another blob, ignore
+						}
+						close(received)
+						return nil
+					}))
+					defer cancel()
+
+					// wait for timeout or the changes register to close the channel
+					select {
+					case <-time.After(30 * time.Second):
+						http.Error(w, "blob wait timeout", http.StatusNotFound)
+						return
+					case <-received:
+						blobReader, err = theBot.BlobStore.Get(blobRef)
+						if err != nil {
+							http.Error(w, "internal error", http.StatusInternalServerError)
+							level.Error(log).Log("msg", "get failure after received", "err", err)
+							return
+						}
+					}
+
+				} else {
+					http.Error(w, err.Error(), http.StatusInternalServerError)
+					level.Error(log).Log("err", err)
+					return
+				}
+			}
+
+			io.Copy(w, blobReader)
+		})
+
+		http.Serve(lis, r)
+	}()
+
 	go func() {
 		for {
 			// Note: This is where the serving starts ;)

go.mod

@@ -5,6 +5,7 @@ go 1.12
 require (
 	github.com/cryptix/go v1.5.0
 	github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b // indirect
+	github.com/gorilla/mux v1.7.3
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/pkg/errors v0.8.1

go.sum

@@ -78,6 +78,8 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
+github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
+github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.1.3/go.mod h1:8KCfur6+4Mqcc6S0FEfKuN15Vl5MgXW92AE8ovaJD0w=
 github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=