Merge pull request #1584 from amangale/manual_order_messaging

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")
 	}
@@ -203,6 +205,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

@@ -65,6 +65,10 @@ type jsonAPIHandler struct {
 	node   *core.OpenBazaarNode
 }
 
+var lastManualScan time.Time
+
+const OfflineMessageScanInterval = 1 * time.Minute
+
 func newJSONAPIHandler(node *core.OpenBazaarNode, authCookie http.Cookie, config schema.APIConfig) *jsonAPIHandler {
 	allowedIPs := make(map[string]bool)
 	for _, ip := range config.AllowedIPs {
@@ -1889,7 +1893,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"`
@@ -1900,7 +1904,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)
 			}
@@ -2824,7 +2828,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)
@@ -2849,7 +2853,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)
 			}
 		}()
@@ -3601,7 +3605,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 {
@@ -3640,7 +3644,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)
 		}
 	}
@@ -4220,3 +4224,75 @@ 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 {
+		// If send message failed, try sending offline message
+		log.Warningf("resending message failed: %v", err)
+		err = i.node.SendOfflineMessage(p, nil, &msg.Msg)
+		if err != nil {
+			log.Errorf("resending offline message failed: %v", err)
+			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 lastManualScan.IsZero() {
+		lastManualScan = time.Now()
+	} else {
+		if time.Since(lastManualScan) >= OfflineMessageScanInterval {
+			i.node.MessageRetriever.RunOnce()
+			lastManualScan = time.Now()
+		}
+	}
+	SanitizedResponse(w, `{}`)
+}