Manually send order messages and trigger offline message scan

api/endpoints.go

@@ -113,6 +113,8 @@ func post(i *jsonAPIHandler, path string, w http.ResponseWriter, r *http.Request
 		i.POSTPost(w, r)
 	case strings.HasPrefix(path, "/ob/bulkupdatecurrency"):
 		i.POSTBulkUpdateCurrency(w, r)
+	case strings.HasPrefix(path, "/ob/resendordermessage"):
+		i.POSTResendOrderMessage(w, r)
 	default:
 		ErrorResponse(w, http.StatusNotFound, "Not Found")
 	}
@@ -201,6 +203,8 @@ func get(i *jsonAPIHandler, path string, w http.ResponseWriter, r *http.Request)
 		i.GETPosts(w, r)
 	case strings.HasPrefix(path, "/ob/post"):
 		i.GETPost(w, r)
+	case strings.HasPrefix(path, "/ob/scanofflinemessages"):
+		i.GETScanOfflineMessages(w, r)
 	default:
 		ErrorResponse(w, http.StatusNotFound, "Not Found")
 	}

api/jsonapi.go

@@ -63,6 +63,8 @@ type jsonAPIHandler struct {
 	node   *core.OpenBazaarNode
 }
 
+var lastManualScan time.Time
+
 func newJSONAPIHandler(node *core.OpenBazaarNode, authCookie http.Cookie, config schema.APIConfig) *jsonAPIHandler {
 	allowedIPs := make(map[string]bool)
 	for _, ip := range config.AllowedIPs {
@@ -1475,7 +1477,7 @@ func (i *jsonAPIHandler) GETProfile(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		if profile.PeerID != peerID {
-			ErrorResponse(w, http.StatusNotFound, err.Error())
+			ErrorResponse(w, http.StatusNotFound, "incorrect peer id")
 			return
 		}
 		w.Header().Set("Cache-Control", "public, max-age=600, immutable")
@@ -1884,7 +1886,7 @@ func (i *jsonAPIHandler) GETModerators(w http.ResponseWriter, r *http.Request) {
 						if err != nil {
 							return
 						}
-						i.node.Broadcast <- repo.PremarshalledNotifier{b}
+						i.node.Broadcast <- repo.PremarshalledNotifier{Payload: b}
 					} else {
 						type wsResp struct {
 							ID     string `json:"id"`
@@ -1895,7 +1897,7 @@ func (i *jsonAPIHandler) GETModerators(w http.ResponseWriter, r *http.Request) {
 						if err != nil {
 							return
 						}
-						i.node.Broadcast <- repo.PremarshalledNotifier{data}
+						i.node.Broadcast <- repo.PremarshalledNotifier{Payload: data}
 					}
 				}(p)
 			}
@@ -2819,7 +2821,7 @@ func (i *jsonAPIHandler) POSTFetchProfiles(w http.ResponseWriter, r *http.Reques
 						if err != nil {
 							return
 						}
-						i.node.Broadcast <- repo.PremarshalledNotifier{ret}
+						i.node.Broadcast <- repo.PremarshalledNotifier{Payload: ret}
 					}
 
 					pro, err := i.node.FetchProfile(pid, useCache)
@@ -2844,7 +2846,7 @@ func (i *jsonAPIHandler) POSTFetchProfiles(w http.ResponseWriter, r *http.Reques
 						respondWithError("error Marshalling to JSON")
 						return
 					}
-					i.node.Broadcast <- repo.PremarshalledNotifier{b}
+					i.node.Broadcast <- repo.PremarshalledNotifier{Payload: b}
 				}(p)
 			}
 		}()
@@ -3594,7 +3596,7 @@ func (i *jsonAPIHandler) POSTFetchRatings(w http.ResponseWriter, r *http.Request
 					if err != nil {
 						return
 					}
-					i.node.Broadcast <- repo.PremarshalledNotifier{ret}
+					i.node.Broadcast <- repo.PremarshalledNotifier{Payload: ret}
 				}
 				ratingBytes, err := ipfs.Cat(i.node.IpfsNode, rid, time.Minute)
 				if err != nil {
@@ -3633,7 +3635,7 @@ func (i *jsonAPIHandler) POSTFetchRatings(w http.ResponseWriter, r *http.Request
 					respondWithError("error marshalling rating")
 					return
 				}
-				i.node.Broadcast <- repo.PremarshalledNotifier{b}
+				i.node.Broadcast <- repo.PremarshalledNotifier{Payload: b}
 			}(r)
 		}
 	}
@@ -4150,3 +4152,65 @@ func (i *jsonAPIHandler) GETPost(w http.ResponseWriter, r *http.Request) {
 	}
 	SanitizedResponseM(w, out, new(pb.SignedPost))
 }
+
+// POSTSendOrderMessage - used to manually send an order message
+func (i *jsonAPIHandler) POSTResendOrderMessage(w http.ResponseWriter, r *http.Request) {
+	type sendRequest struct {
+		OrderID     string `json:"orderID"`
+		MessageType string `json:"messageType"`
+	}
+
+	var args sendRequest
+	decoder := json.NewDecoder(r.Body)
+	err := decoder.Decode(&args)
+	if err != nil {
+		ErrorResponse(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	if args.OrderID == "" {
+		ErrorResponse(w, http.StatusBadRequest, core.ErrOrderNotFound.Error())
+		return
+	}
+
+	var msgType int32
+	var ok bool
+
+	if msgType, ok = pb.Message_MessageType_value[args.MessageType]; !ok {
+		ErrorResponse(w, http.StatusBadRequest, "invalid order message type")
+		return
+	}
+
+	msg, peerID, err := i.node.Datastore.Messages().
+		GetByOrderIDType(args.OrderID, pb.Message_MessageType(msgType))
+	if err != nil || msg == nil || msg.Msg.GetPayload() == nil {
+		ErrorResponse(w, http.StatusBadRequest, "order message not found")
+		return
+	}
+
+	p, err := peer.IDB58Decode(peerID)
+	if err != nil {
+		ErrorResponse(w, http.StatusBadRequest, "invalid peer id")
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	err = i.node.Service.SendMessage(ctx, p, &msg.Msg)
+	if err != nil {
+		ErrorResponse(w, http.StatusBadRequest, "order message not sent")
+		return
+	}
+
+	SanitizedResponse(w, "")
+}
+
+// GETScanOfflineMessages - used to manually trigger offline message scan
+func (i *jsonAPIHandler) GETScanOfflineMessages(w http.ResponseWriter, r *http.Request) {
+	if time.Since(lastManualScan).Minutes() > 10 {
+		i.node.MessageRetriever.RunOnce()
+		lastManualScan = time.Now()
+	}
+	SanitizedResponse(w, "")
+}

core/net.go

@@ -2,17 +2,18 @@ package core
 
 import (
 	"errors"
+	"fmt"
+	"sync"
+	"time"
 
 	libp2p "gx/ipfs/QmTW4SdgBWq9GjsBsHeUx8WuGxzhgzAf88UMH2w62PC8yK/go-libp2p-crypto"
 	"gx/ipfs/QmTbxNB1NwDesLmKTscr4udL2tVP7MaxvXnD1D9yX7g3PN/go-cid"
 	"gx/ipfs/QmYVXrKrKHDC9FobgmcmshCDyWwdrfwfanNQN4oxJ9Fk3h/go-libp2p-peer"
 	"gx/ipfs/QmerPMzPk1mJVowm8KgmoknWa4yCYvvugMPsgWmDNUvDLW/go-multihash"
 
-	"sync"
-	"time"
-
 	"github.com/OpenBazaar/openbazaar-go/ipfs"
 	"github.com/OpenBazaar/openbazaar-go/pb"
+	"github.com/OpenBazaar/openbazaar-go/repo"
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/any"
@@ -276,6 +277,11 @@ func (n *OpenBazaarNode) SendOrder(peerID string, contract *pb.RicardianContract
 		Payload:     pbAny,
 	}
 
+	orderID0, _ := n.CalcOrderID(contract.BuyerOrder)
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", orderID0, int(pb.Message_ORDER)),
+		orderID0, pb.Message_ORDER, peerID, repo.Message{Msg: m})
+
 	resp, err = n.Service.SendRequest(ctx, p, &m)
 	if err != nil {
 		return resp, err
@@ -302,6 +308,11 @@ func (n *OpenBazaarNode) SendOrderConfirmation(peerID string, contract *pb.Ricar
 	if err != nil {
 		return err
 	}
+	orderID0, _ := n.CalcOrderID(contract.BuyerOrder)
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", orderID0, int(pb.Message_ORDER_CONFIRMATION)),
+		orderID0, pb.Message_ORDER_CONFIRMATION, peerID, repo.Message{Msg: m})
+
 	return n.sendMessage(peerID, &k, m)
 }
 
@@ -324,6 +335,9 @@ func (n *OpenBazaarNode) SendCancel(peerID, orderID string) error {
 		}
 		kp = &k
 	}
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", orderID, int(pb.Message_ORDER_CANCEL)),
+		orderID, pb.Message_ORDER_CANCEL, peerID, repo.Message{Msg: m})
 	return n.sendMessage(peerID, kp, m)
 }
 
@@ -349,6 +363,9 @@ func (n *OpenBazaarNode) SendReject(peerID string, rejectMessage *pb.OrderReject
 		}
 		kp = &k
 	}
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", rejectMessage.OrderID, int(pb.Message_ORDER_REJECT)),
+		rejectMessage.OrderID, pb.Message_ORDER_REJECT, peerID, repo.Message{Msg: m})
 	return n.sendMessage(peerID, kp, m)
 }
 
@@ -379,6 +396,10 @@ func (n *OpenBazaarNode) SendOrderFulfillment(peerID string, k *libp2p.PubKey, f
 		MessageType: pb.Message_ORDER_FULFILLMENT,
 		Payload:     a,
 	}
+	orderID0, _ := n.CalcOrderID(fulfillmentMessage.BuyerOrder)
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", orderID0, int(pb.Message_ORDER_FULFILLMENT)),
+		orderID0, pb.Message_ORDER_FULFILLMENT, peerID, repo.Message{Msg: m})
 	return n.sendMessage(peerID, k, m)
 }
 
@@ -395,6 +416,10 @@ func (n *OpenBazaarNode) SendOrderCompletion(peerID string, k *libp2p.PubKey, co
 	if err != nil {
 		return err
 	}
+	orderID0, _ := n.CalcOrderID(completionMessage.BuyerOrder)
+	n.Datastore.Messages().Put(
+		fmt.Sprintf("%s-%d", orderID0, int(pb.Message_ORDER_COMPLETION)),
+		orderID0, pb.Message_ORDER_COMPLETION, peerID, repo.Message{Msg: m})
 	return n.sendMessage(peerID, k, m)
 }
 

net/retriever/retriever.go

@@ -118,6 +118,12 @@ func (m *MessageRetriever) Run() {
 	}
 }
 
+// RunOnce - used to fetch messages only once
+func (m *MessageRetriever) RunOnce() {
+	go m.fetchPointers(true)
+	go m.fetchPointers(false)
+}
+
 func (m *MessageRetriever) fetchPointers(useDHT bool) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

repo/datastore.go

@@ -28,6 +28,7 @@ type Datastore interface {
 	Coupons() CouponStore
 	TxMetadata() TransactionMetadataStore
 	ModeratedStores() ModeratedStore
+	Messages() MessageStore
 	Ping() error
 	Close()
 }
@@ -431,3 +432,17 @@ type WatchedScriptStore interface {
 	Queryable
 	wallet.WatchedScripts
 }
+
+// MessageStore is the messages table interface
+type MessageStore interface {
+	Queryable
+
+	// Save a new message
+	Put(messageID, orderID string, mType pb.Message_MessageType, peerID string, msg Message) error
+
+	// GetByOrderIDType returns the message for specified order and type
+	GetByOrderIDType(orderID string, mType pb.Message_MessageType) (*Message, string, error)
+
+	// GetByMessageIDType returns the message for specified message id
+	GetByMessageIDType(messageID string) (*Message, string, error)
+}

repo/db/db.go

@@ -36,6 +36,7 @@ type SQLiteDatastore struct {
 	coupons         repo.CouponStore
 	txMetadata      repo.TransactionMetadataStore
 	moderatedStores repo.ModeratedStore
+	messages        repo.MessageStore
 	db              *sql.DB
 	lock            *sync.Mutex
 }
@@ -81,6 +82,7 @@ func NewSQLiteDatastore(db *sql.DB, l *sync.Mutex, coinType wallet.CoinType) *SQ
 		coupons:         NewCouponStore(db, l),
 		txMetadata:      NewTransactionMetadataStore(db, l),
 		moderatedStores: NewModeratedStore(db, l),
+		messages:        NewMessageStore(db, l),
 		db:              db,
 		lock:            l,
 	}
@@ -183,6 +185,11 @@ func (d *SQLiteDatastore) ModeratedStores() repo.ModeratedStore {
 	return d.moderatedStores
 }
 
+// Messages - return the messages datastore
+func (d *SQLiteDatastore) Messages() repo.MessageStore {
+	return d.messages
+}
+
 func (d *SQLiteDatastore) Copy(dbPath string, password string) error {
 	d.lock.Lock()
 	defer d.lock.Unlock()