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,11 +26,12 @@ 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"
 
-	"github.com/pietgeursen/gobotexample/internal/multiserver"
+	"github.com/sunrise-choir/sunrise-social-gobot/internal/multiserver"
 )
 
 var (
@@ -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 ;)

bot_test.go

@@ -8,8 +8,8 @@ import (
 
 	"go.cryptoscope.co/margaret"
 
-	"github.com/pietgeursen/gobotexample"
 	"github.com/stretchr/testify/require"
+	gobotexample "github.com/sunrise-choir/sunrise-social-gobot"
 )
 
 func TestStartStop(t *testing.T) {
@@ -43,7 +43,7 @@ func TestPublish(t *testing.T) {
 	r.NoError(err)
 	r.EqualValues(margaret.SeqEmpty, c)
 
-	err = gobotexample.Publish(`{"type":"test", "hello":"piet"}`, nil)
+	err = gobotexample.Publish(`{"type":"test", "hello":"piet"}`)
 	r.NoError(err)
 
 	c, err = gobotexample.CurrentMessageCount()
@@ -59,7 +59,7 @@ func TestPublish(t *testing.T) {
 	r.NoError(err)
 	r.EqualValues(len(v), 1)
 
-	err = gobotexample.Publish(`{"type":"another", "hello":"piet!!!"}`, nil)
+	err = gobotexample.Publish(`{"type":"another", "hello":"piet!!!"}`)
 	r.NoError(err)
 
 	c, err = gobotexample.CurrentMessageCount()

go.mod

@@ -1,12 +1,12 @@
-module github.com/pietgeursen/gobotexample
+module github.com/sunrise-choir/sunrise-social-gobot
 
 go 1.12
 
 require (
 	github.com/cryptix/go v1.5.0
 	github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b // indirect
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v1.0.1 // indirect
+	github.com/go-kit/kit v0.9.0
+	github.com/gorilla/mux v1.7.3
 	github.com/pkg/errors v0.8.1
 	github.com/stretchr/testify v1.4.0
 	go.cryptoscope.co/luigi v0.3.5-0.20190924074117-8ca146aad481