Merge lp:~pedronis/ubuntu-push/unicast-endp into lp:ubuntu-push/automatic

[Samuele Pedroni]

/notify api endpoint for unicast, takes explicit user id,device id pair for now

/notify api endpoint for unicast, takes explicit user id,device id pair for now

[John Lenton (chipaca)]

review: Approve

[lp:~pedronis/ubuntu-push/unicast-endp]

145. By Samuele Pedroni on 2014-05-02

Merged unicast-broker into unicast-endp.

=== modified file 'Makefile'
--- Makefile	2014-05-01 10:24:23 +0000
+++ Makefile	2014-05-02 15:13:47 +0000
@@ -11,6 +11,7 @@
 GODEPS += launchpad.net/go-dbus/v1
 GODEPS += launchpad.net/go-xdg/v0
 GODEPS += code.google.com/p/gosqlite/sqlite3
+GODEPS += launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid/uuid
 
 TOTEST = $(shell env GOPATH=$(GOPATH) go list $(PROJECT)/...|grep -v acceptance|grep -v http13client )
 

=== modified file 'dependencies.tsv'
--- dependencies.tsv	2014-04-30 16:44:16 +0000
+++ dependencies.tsv	2014-05-02 15:13:47 +0000
@@ -2,3 +2,4 @@
 launchpad.net/go-dbus/v1	bzr	[e-mail]-20140206110213-pbzcr6ucaz3rqmnw	125
 launchpad.net/go-xdg/v0	bzr	[e-mail]-20140208094800-gubd5md7cro3mtxa	10
 launchpad.net/gocheck	bzr	[e-mail]-20140127131816-zshobk1qqme626xw	86
+launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid	bzr	[e-mail]-20140130122455-pm9h8etl4owp90lg	1

=== modified file 'server/api/handlers.go'
--- server/api/handlers.go	2014-02-20 17:09:03 +0000
+++ server/api/handlers.go	2014-05-02 15:13:47 +0000
@@ -19,12 +19,15 @@
 package api
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"time"
 
+	"launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid/uuid"
+
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/store"
@@ -93,6 +96,11 @@
 		ioError,
 		"Could not read request body",
 	}
+	ErrMissingIdField = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Missing id field",
+	}
 	ErrMissingData = &APIError{
 		http.StatusBadRequest,
 		invalidRequest,
@@ -130,10 +138,17 @@
 	}
 )
 
-type Message struct {
-	Registration string          `json:"registration"`
-	CoalesceTag  string          `json:"coalesce_tag"`
-	Data         json.RawMessage `json:"data"`
+type castCommon struct {
+}
+
+type Unicast struct {
+	UserId   string `json:"userid"`
+	DeviceId string `json:"deviceid"`
+	AppId    string `json:"appid"`
+	//Registration string          `json:"registration"`
+	//CoalesceTag  string          `json:"coalesce_tag"`
+	ExpireOn string          `json:"expire_on"`
+	Data     json.RawMessage `json:"data"`
 }
 
 // Broadcast request JSON object.
@@ -198,11 +213,11 @@
 
 var zeroTime = time.Time{}
 
-func checkBroadcast(bcast *Broadcast) (time.Time, *APIError) {
-	if len(bcast.Data) == 0 {
+func checkCastCommon(data json.RawMessage, expireOn string) (time.Time, *APIError) {
+	if len(data) == 0 {
 		return zeroTime, ErrMissingData
 	}
-	expire, err := time.Parse(time.RFC3339, bcast.ExpireOn)
+	expire, err := time.Parse(time.RFC3339, expireOn)
 	if err != nil {
 		return zeroTime, ErrInvalidExpiration
 	}
@@ -212,6 +227,10 @@
 	return expire, nil
 }
 
+func checkBroadcast(bcast *Broadcast) (time.Time, *APIError) {
+	return checkCastCommon(bcast.Data, bcast.ExpireOn)
+}
+
 type StoreForRequest func(w http.ResponseWriter, request *http.Request) (store.PendingStore, error)
 
 // context holds the interfaces to delegate to serving requests
@@ -234,6 +253,20 @@
 	return sto, nil
 }
 
+func (ctx *context) prepare(w http.ResponseWriter, request *http.Request, reqObj interface{}) (store.PendingStore, *APIError) {
+	body, apiErr := ReadBody(request, MaxRequestBodyBytes)
+	if apiErr != nil {
+		return nil, apiErr
+	}
+
+	err := json.Unmarshal(body, reqObj)
+	if err != nil {
+		return nil, ErrMalformedJSONObject
+	}
+
+	return ctx.getStore(w, request)
+}
+
 type BroadcastHandler struct {
 	*context
 }
@@ -270,23 +303,13 @@
 		}
 	}()
 
-	body, apiErr := ReadBody(request, MaxRequestBodyBytes)
-	if apiErr != nil {
-		return
-	}
-
-	sto, apiErr := h.getStore(writer, request)
-	if apiErr != nil {
-		return
-	}
-	defer sto.Close()
-
 	broadcast := &Broadcast{}
-	err := json.Unmarshal(body, broadcast)
-	if err != nil {
-		apiErr = ErrMalformedJSONObject
+
+	sto, apiErr := h.prepare(writer, request, broadcast)
+	if apiErr != nil {
 		return
 	}
+	defer sto.Close()
 
 	apiErr = h.doBroadcast(sto, broadcast)
 	if apiErr != nil {
@@ -297,6 +320,64 @@
 	fmt.Fprintf(writer, `{"ok":true}`)
 }
 
+type UnicastHandler struct {
+	*context
+}
+
+func checkUnicast(ucast *Unicast) (time.Time, *APIError) {
+	if ucast.UserId == "" || ucast.DeviceId == "" || ucast.AppId == "" {
+		return zeroTime, ErrMissingIdField
+	}
+	return checkCastCommon(ucast.Data, ucast.ExpireOn)
+}
+
+// use a base64 encoded TimeUUID
+var generateMsgId = func() string {
+	return base64.StdEncoding.EncodeToString(uuid.NewUUID())
+}
+
+func (h *UnicastHandler) doUnicast(sto store.PendingStore, ucast *Unicast) *APIError {
+	expire, apiErr := checkUnicast(ucast)
+	if apiErr != nil {
+		return apiErr
+	}
+	chanId := store.UnicastInternalChannelId(ucast.UserId, ucast.DeviceId)
+	msgId := generateMsgId()
+	err := sto.AppendToUnicastChannel(chanId, ucast.AppId, ucast.Data, msgId, expire)
+	if err != nil {
+		h.logger.Errorf("could not store notification: %v", err)
+		return ErrCouldNotStoreNotification
+	}
+
+	h.broker.Unicast(chanId)
+	return nil
+}
+
+func (h *UnicastHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	var apiErr *APIError
+	defer func() {
+		if apiErr != nil {
+			RespondError(writer, apiErr)
+		}
+	}()
+
+	unicast := &Unicast{}
+
+	sto, apiErr := h.prepare(writer, request, unicast)
+	if apiErr != nil {
+		return
+	}
+	defer sto.Close()
+
+	apiErr = h.doUnicast(sto, unicast)
+	if apiErr != nil {
+		return
+	}
+
+	writer.Header().Set("Content-Type", "application/json")
+	fmt.Fprintf(writer, `{"ok":true}`)
+}
+
 // MakeHandlersMux makes a handler that dispatches for the various API endpoints.
 func MakeHandlersMux(storeForRequest StoreForRequest, broker broker.BrokerSending, logger logger.Logger) *http.ServeMux {
 	ctx := &context{
@@ -306,5 +387,6 @@
 	}
 	mux := http.NewServeMux()
 	mux.Handle("/broadcast", &BroadcastHandler{context: ctx})
+	mux.Handle("/notify", &UnicastHandler{context: ctx})
 	return mux
 }

=== modified file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	2014-05-02 15:13:47 +0000
+++ server/api/handlers_test.go	2014-05-02 15:13:47 +0000
@@ -18,6 +18,7 @@
 
 import (
 	"bytes"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -32,7 +33,7 @@
 
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/store"
-	helpers "launchpad.net/ubuntu-push/testing"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 func TestHandlers(t *testing.T) { TestingT(t) }
@@ -42,14 +43,14 @@
 	json            string
 	client          *http.Client
 	c               *C
-	testlog         *helpers.TestLogger
+	testlog         *help.TestLogger
 }
 
 var _ = Suite(&handlersSuite{})
 
 func (s *handlersSuite) SetUpTest(c *C) {
 	s.client = &http.Client{}
-	s.testlog = helpers.NewTestLogger(c, "error")
+	s.testlog = help.NewTestLogger(c, "error")
 }
 
 func (s *handlersSuite) TestAPIError(c *C) {
@@ -99,7 +100,7 @@
 
 var future = time.Now().Add(4 * time.Hour).Format(time.RFC3339)
 
-func (s *handlersSuite) TestCheckBroadcast(c *C) {
+func (s *handlersSuite) TestCheckCastBroadcastAndCommon(c *C) {
 	payload := json.RawMessage(`{"foo":"bar"}`)
 	broadcast := &Broadcast{
 		Channel:  "system",
@@ -135,11 +136,11 @@
 }
 
 type checkBrokerSending struct {
-	store    store.PendingStore
-	chanId   store.InternalChannelId
-	err      error
-	top      int64
-	payloads []json.RawMessage
+	store         store.PendingStore
+	chanId        store.InternalChannelId
+	err           error
+	top           int64
+	notifications []protocol.Notification
 }
 
 func (cbsend *checkBrokerSending) Broadcast(chanId store.InternalChannelId) {
@@ -147,11 +148,15 @@
 	cbsend.err = err
 	cbsend.chanId = chanId
 	cbsend.top = top
-	cbsend.payloads = protocol.ExtractPayloads(notifications)
+	cbsend.notifications = notifications
 }
 
 func (cbsend *checkBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
-	// xxx later
+	// for now
+	if len(chanIds) != 1 {
+		panic("not expecting many chan ids for now")
+	}
+	cbsend.Broadcast(chanIds[0])
 }
 
 func (s *handlersSuite) TestDoBroadcast(c *C) {
@@ -168,7 +173,7 @@
 	c.Check(bsend.err, IsNil)
 	c.Check(bsend.chanId, Equals, store.SystemInternalChannelId)
 	c.Check(bsend.top, Equals, int64(1))
-	c.Check(bsend.payloads, DeepEquals, []json.RawMessage{payload})
+	c.Check(bsend.notifications, DeepEquals, help.Ns(payload))
 }
 
 func (s *handlersSuite) TestDoBroadcastUnknownChannel(c *C) {
@@ -197,6 +202,11 @@
 	return isto.intercept("AppendToChannel", err)
 }
 
+func (isto *interceptInMemoryPendingStore) AppendToUnicastChannel(chanId store.InternalChannelId, appId string, payload json.RawMessage, msgId string, expiration time.Time) error {
+	err := isto.InMemoryPendingStore.AppendToUnicastChannel(chanId, appId, payload, msgId, expiration)
+	return isto.intercept("AppendToUnicastChannel", err)
+}
+
 func (s *handlersSuite) TestDoBroadcastUnknownError(c *C) {
 	sto := &interceptInMemoryPendingStore{
 		store.NewInMemoryPendingStore(),
@@ -234,6 +244,115 @@
 	c.Check(s.testlog.Captured(), Equals, "ERROR could not store notification: fail\n")
 }
 
+func (s *handlersSuite) TestCheckUnicast(c *C) {
+	payload := json.RawMessage(`{"foo":"bar"}`)
+	unicast := func() *Unicast {
+		return &Unicast{
+			UserId:   "user1",
+			DeviceId: "DEV1",
+			AppId:    "app1",
+			ExpireOn: future,
+			Data:     payload,
+		}
+	}
+	u := unicast()
+	expire, apiErr := checkUnicast(u)
+	c.Assert(apiErr, IsNil)
+	c.Check(expire.Format(time.RFC3339), Equals, future)
+
+	u = unicast()
+	u.UserId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.AppId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.DeviceId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.Data = json.RawMessage(nil)
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingData)
+}
+
+func (s *handlersSuite) TestGenerateMsgId(c *C) {
+	msgId := generateMsgId()
+	decoded, err := base64.StdEncoding.DecodeString(msgId)
+	c.Assert(err, IsNil)
+	c.Check(decoded, HasLen, 16)
+}
+
+func (s *handlersSuite) TestDoUnicast(c *C) {
+	prevGenMsgId := generateMsgId
+	defer func() {
+		generateMsgId = prevGenMsgId
+	}()
+	generateMsgId = func() string {
+		return "MSG-ID"
+	}
+	sto := store.NewInMemoryPendingStore()
+	bsend := &checkBrokerSending{store: sto}
+	bh := &UnicastHandler{&context{nil, bsend, nil}}
+	payload := json.RawMessage(`{"a": 1}`)
+	apiErr := bh.doUnicast(sto, &Unicast{
+		UserId:   "user1",
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     payload,
+	})
+	c.Check(apiErr, IsNil)
+	c.Check(bsend.err, IsNil)
+	c.Check(bsend.chanId, Equals, store.UnicastInternalChannelId("user1", "DEV1"))
+	c.Check(bsend.top, Equals, int64(0))
+	c.Check(bsend.notifications, DeepEquals, []protocol.Notification{
+		protocol.Notification{
+			AppId:   "app1",
+			MsgId:   "MSG-ID",
+			Payload: payload,
+		},
+	})
+}
+
+func (s *handlersSuite) TestDoUnicastMissingIdField(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	bh := &UnicastHandler{}
+	apiErr := bh.doUnicast(sto, &Unicast{
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrMissingIdField)
+}
+
+func (s *handlersSuite) TestDoUnicastCouldNotStoreNotification(c *C) {
+	sto := &interceptInMemoryPendingStore{
+		store.NewInMemoryPendingStore(),
+		func(meth string, err error) error {
+			if meth == "AppendToUnicastChannel" {
+				return errors.New("fail")
+			}
+			return err
+		},
+	}
+	ctx := &context{logger: s.testlog}
+	bh := &UnicastHandler{ctx}
+	apiErr := bh.doUnicast(sto, &Unicast{
+		UserId:   "user1",
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrCouldNotStoreNotification)
+	c.Check(s.testlog.Captured(), Equals, "ERROR could not store notification: fail\n")
+}
+
 func newPostRequest(path string, message interface{}, server *httptest.Server) *http.Request {
 	packedMessage, err := json.Marshal(message)
 	if err != nil {
@@ -274,7 +393,11 @@
 }
 
 func (bsend testBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
-	// xxx later
+	// for now
+	if len(chanIds) != 1 {
+		panic("not expecting many chan ids for now")
+	}
+	bsend.chanId <- chanIds[0]
 }
 
 func (s *handlersSuite) TestRespondsToBasicSystemBroadcast(c *C) {
@@ -489,3 +612,73 @@
 
 	checkError(c, response, ErrWrongRequestMethod)
 }
+
+func (s *handlersSuite) TestRespondsUnicast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return sto, nil
+	}
+	bsend := testBrokerSending{make(chan store.InternalChannelId, 1)}
+	testServer := httptest.NewServer(MakeHandlersMux(stoForReq, bsend, nil))
+	defer testServer.Close()
+
+	payload := json.RawMessage(`{"foo":"bar"}`)
+
+	request := newPostRequest("/notify", &Unicast{
+		UserId:   "user2",
+		DeviceId: "dev3",
+		AppId:    "app2",
+		ExpireOn: future,
+		Data:     payload,
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+
+	c.Check(response.StatusCode, Equals, http.StatusOK)
+	c.Check(response.Header.Get("Content-Type"), Equals, "application/json")
+	body, err := getResponseBody(response)
+	c.Assert(err, IsNil)
+	c.Check(string(body), Matches, ".*ok.*")
+
+	chanId := store.UnicastInternalChannelId("user2", "dev3")
+	c.Check(<-bsend.chanId, Equals, chanId)
+	top, notifications, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(top, Equals, int64(0))
+	c.Check(notifications, HasLen, 1)
+}
+
+func (s *handlersSuite) TestCannotUnicastTooBigMessages(c *C) {
+	testServer := httptest.NewServer(&UnicastHandler{})
+	defer testServer.Close()
+
+	bigString := strings.Repeat("a", MaxRequestBodyBytes)
+	dataString := fmt.Sprintf(`"%v"`, bigString)
+
+	request := newPostRequest("/", &Unicast{
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrRequestBodyTooLarge)
+}
+
+func (s *handlersSuite) TestCannotUnicastWithMissingFields(c *C) {
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return store.NewInMemoryPendingStore(), nil
+	}
+	ctx := &context{stoForReq, nil, nil}
+	testServer := httptest.NewServer(&UnicastHandler{ctx})
+	defer testServer.Close()
+
+	request := newPostRequest("/", &Unicast{
+		Data: json.RawMessage(`{"foo":"bar"}`),
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrMissingIdField)
+}