Merge lp:~pedronis/ubuntu-push/broadcast-expiration into lp:ubuntu-push

[Samuele Pedroni]

add support for expiration of broadcasts with trivial implementation in the in-memory pending store

add support for expiration of broadcasts with trivial implementation in the in-memory pending store,

acceptance test coming in a later branch

[lp:~pedronis/ubuntu-push/broadcast-expiration]

61. By Samuele Pedroni on 2014-02-10

fix comment

62. By Samuele Pedroni on 2014-02-10

formatting

63. By Samuele Pedroni on 2014-02-10

fix acceptance test

64. By Samuele Pedroni on 2014-02-10

merge reorg some imports

65. By Samuele Pedroni on 2014-02-10

Merged reorg-imports into expiration-flex-acceptance.

66. By Samuele Pedroni on 2014-02-10

Merged fix-tweak-tests into expiration-flex-acceptance.

[John Lenton (chipaca)]

What's the plan for exposing this via REST?

review: Approve

[John Lenton (chipaca)]

Stupid question, it's right there. Sorry!

=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-02-10 23:38:19 +0000
+++ server/acceptance/acceptance_test.go	2014-02-10 23:46:49 +0000
@@ -301,6 +301,8 @@
 
 // Tests about broadcast
 
+var future = time.Now().Add(9 * time.Hour).Format(time.RFC3339)
+
 func (s *acceptanceSuite) postRequest(path string, message interface{}) (string, error) {
 	packedMessage, err := json.Marshal(message)
 	if err != nil {
@@ -351,8 +353,9 @@
 	}
 	events, errCh := s.startClient(c, "DEVB", intercept, nil)
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"n": 42}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"n": 42}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -365,8 +368,9 @@
 func (s *acceptanceSuite) TestBroadcastPending(c *C) {
 	// send broadcast that will be pending
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -393,8 +397,9 @@
 	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
 	for i := 0; i < 32; i++ {
 		got, err := s.postRequest("/broadcast", &api.Broadcast{
-			Channel: "system",
-			Data:    json.RawMessage(fmt.Sprintf(payloadFmt, i)),
+			Channel:  "system",
+			ExpireOn: future,
+			Data:     json.RawMessage(fmt.Sprintf(payloadFmt, i)),
 		})
 		c.Assert(err, IsNil)
 		c.Assert(got, Matches, ".*ok.*")
@@ -434,8 +439,9 @@
 	events2, errCh2 := s.startClient(c, "DEV2", intercept, nil)
 	// broadcast
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"n": 42}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"n": 42}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -460,8 +466,9 @@
 	}
 	events, errCh := s.startClient(c, "DEVD", intercept, nil)
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -471,8 +478,9 @@
 	c.Check(len(errCh), Equals, 0)
 	// another broadcast
 	got, err = s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 2}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -490,14 +498,16 @@
 func (s *acceptanceSuite) TestBroadcastTooAhead(c *C) {
 	// send broadcasts that will be pending
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	got, err = s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 2}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -545,14 +555,16 @@
 func (s *acceptanceSuite) TestBroadcastWayBehind(c *C) {
 	// send broadcasts that will be pending
 	got, err := s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	got, err = s.postRequest("/broadcast", &api.Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"b": 2}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"b": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")

=== modified file 'server/api/handlers.go'
--- server/api/handlers.go	2014-02-10 23:19:08 +0000
+++ server/api/handlers.go	2014-02-10 23:46:49 +0000
@@ -23,6 +23,7 @@
 	"fmt"
 	"io"
 	"net/http"
+	"time"
 
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/server/broker"
@@ -97,6 +98,16 @@
 		invalidRequest,
 		"Missing data field",
 	}
+	ErrInvalidExpiration = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Invalid expiration date",
+	}
+	ErrPastExpiration = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Past expiration date",
+	}
 	ErrUnknownChannel = &APIError{
 		http.StatusBadRequest,
 		unknownChannel,
@@ -122,9 +133,9 @@
 
 // Broadcast request JSON object.
 type Broadcast struct {
-	Channel     string          `json:"channel"`
-	ExpireAfter uint8           `json:"expire_after"`
-	Data        json.RawMessage `json:"data"`
+	Channel  string          `json:"channel"`
+	ExpireOn string          `json:"expire_on"`
+	Data     json.RawMessage `json:"data"`
 }
 
 func respondError(writer http.ResponseWriter, apiErr *APIError) {
@@ -178,11 +189,20 @@
 	return body, nil
 }
 
-func checkBroadcast(bcast *Broadcast) *APIError {
+var zeroTime = time.Time{}
+
+func checkBroadcast(bcast *Broadcast) (time.Time, *APIError) {
 	if len(bcast.Data) == 0 {
-		return ErrMissingData
-	}
-	return nil
+		return zeroTime, ErrMissingData
+	}
+	expire, err := time.Parse(time.RFC3339, bcast.ExpireOn)
+	if err != nil {
+		return zeroTime, ErrInvalidExpiration
+	}
+	if expire.Before(time.Now()) {
+		return zeroTime, ErrPastExpiration
+	}
+	return expire, nil
 }
 
 // state holds the interfaces to delegate to serving requests
@@ -195,7 +215,7 @@
 type BroadcastHandler state
 
 func (h *BroadcastHandler) doBroadcast(bcast *Broadcast) *APIError {
-	apiErr := checkBroadcast(bcast)
+	expire, apiErr := checkBroadcast(bcast)
 	if apiErr != nil {
 		return apiErr
 	}
@@ -208,8 +228,7 @@
 			return ErrUnknown
 		}
 	}
-	// xxx ignoring expiration for now
-	err = h.store.AppendToChannel(chanId, bcast.Data)
+	err = h.store.AppendToChannel(chanId, bcast.Data, expire)
 	if err != nil {
 		// assume this for now
 		return ErrCouldNotStoreNotification

=== modified file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	2014-02-10 23:19:08 +0000
+++ server/api/handlers_test.go	2014-02-10 23:46:49 +0000
@@ -26,6 +26,7 @@
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	. "launchpad.net/gocheck"
 
@@ -65,6 +66,43 @@
 	c.Check(err, Equals, ErrCouldNotReadBody)
 }
 
+var future = time.Now().Add(4 * time.Hour).Format(time.RFC3339)
+
+func (s *handlersSuite) TestCheckBroadcast(c *C) {
+	payload := json.RawMessage(`{"foo":"bar"}`)
+	broadcast := &Broadcast{
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     payload,
+	}
+	expire, err := checkBroadcast(broadcast)
+	c.Check(err, IsNil)
+	c.Check(expire.Format(time.RFC3339), Equals, future)
+
+	broadcast = &Broadcast{
+		Channel:  "system",
+		ExpireOn: future,
+	}
+	_, err = checkBroadcast(broadcast)
+	c.Check(err, Equals, ErrMissingData)
+
+	broadcast = &Broadcast{
+		Channel:  "system",
+		ExpireOn: "12:00",
+		Data:     payload,
+	}
+	_, err = checkBroadcast(broadcast)
+	c.Check(err, Equals, ErrInvalidExpiration)
+
+	broadcast = &Broadcast{
+		Channel:  "system",
+		ExpireOn: time.Now().Add(-10 * time.Hour).Format(time.RFC3339),
+		Data:     payload,
+	}
+	_, err = checkBroadcast(broadcast)
+	c.Check(err, Equals, ErrPastExpiration)
+}
+
 type checkBrokerSending struct {
 	store    store.PendingStore
 	chanId   store.InternalChannelId
@@ -87,8 +125,9 @@
 	bh := &BroadcastHandler{sto, bsend, nil}
 	payload := json.RawMessage(`{"a": 1}`)
 	apiErr := bh.doBroadcast(&Broadcast{
-		Channel: "system",
-		Data:    payload,
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     payload,
 	})
 	c.Check(apiErr, IsNil)
 	c.Check(bsend.err, IsNil)
@@ -101,8 +140,9 @@
 	sto := store.NewInMemoryPendingStore()
 	bh := &BroadcastHandler{sto, nil, nil}
 	apiErr := bh.doBroadcast(&Broadcast{
-		Channel: "unknown",
-		Data:    json.RawMessage(`{"a": 1}`),
+		Channel:  "unknown",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
 	})
 	c.Check(apiErr, Equals, ErrUnknownChannel)
 }
@@ -117,8 +157,8 @@
 	return chanId, isto.intercept("GetInternalChannelId", err)
 }
 
-func (isto *interceptInMemoryPendingStore) AppendToChannel(chanId store.InternalChannelId, payload json.RawMessage) error {
-	err := isto.InMemoryPendingStore.AppendToChannel(chanId, payload)
+func (isto *interceptInMemoryPendingStore) AppendToChannel(chanId store.InternalChannelId, payload json.RawMessage, expiration time.Time) error {
+	err := isto.InMemoryPendingStore.AppendToChannel(chanId, payload, expiration)
 	return isto.intercept("AppendToChannel", err)
 }
 
@@ -131,8 +171,9 @@
 	}
 	bh := &BroadcastHandler{sto, nil, nil}
 	apiErr := bh.doBroadcast(&Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"a": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
 	})
 	c.Check(apiErr, Equals, ErrUnknown)
 }
@@ -149,8 +190,9 @@
 	}
 	bh := &BroadcastHandler{sto, nil, nil}
 	apiErr := bh.doBroadcast(&Broadcast{
-		Channel: "system",
-		Data:    json.RawMessage(`{"a": 1}`),
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
 	})
 	c.Check(apiErr, Equals, ErrCouldNotStoreNotification)
 }
@@ -203,9 +245,9 @@
 	payload := json.RawMessage(`{"foo":"bar"}`)
 
 	request := newPostRequest("/broadcast", &Broadcast{
-		Channel:     "system",
-		ExpireAfter: 60,
-		Data:        payload,
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     payload,
 	}, testServer)
 
 	response, err := s.client.Do(request)
@@ -234,9 +276,9 @@
 	payload := json.RawMessage(`{"foo":"bar"}`)
 
 	request := newPostRequest("/broadcast", &Broadcast{
-		Channel:     "unknown",
-		ExpireAfter: 60,
-		Data:        payload,
+		Channel:  "unknown",
+		ExpireOn: future,
+		Data:     payload,
 	}, testServer)
 
 	response, err := s.client.Do(request)
@@ -286,9 +328,9 @@
 	dataString := fmt.Sprintf(`"%v"`, bigString)
 
 	request := newPostRequest("/", &Broadcast{
-		Channel:     "some-channel",
-		ExpireAfter: 60,
-		Data:        json.RawMessage([]byte(dataString)),
+		Channel:  "some-channel",
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
 	}, testServer)
 
 	response, err := s.client.Do(request)
@@ -303,9 +345,9 @@
 	dataString := `{"foo":"bar"}`
 
 	request := newPostRequest("/", &Broadcast{
-		Channel:     "some-channel",
-		ExpireAfter: 60,
-		Data:        json.RawMessage([]byte(dataString)),
+		Channel:  "some-channel",
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
 	}, testServer)
 	request.ContentLength = -1
 
@@ -338,9 +380,9 @@
 	dataString := `{"foo":"bar"}`
 
 	request := newPostRequest("/", &Broadcast{
-		Channel:     "some-channel",
-		ExpireAfter: 60,
-		Data:        json.RawMessage([]byte(dataString)),
+		Channel:  "some-channel",
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
 	}, testServer)
 	request.Header.Set("Content-Type", "text/plain")
 
@@ -355,9 +397,9 @@
 
 	dataString := `{"foo":"bar"}`
 	packedMessage, err := json.Marshal(&Broadcast{
-		Channel:     "some-channel",
-		ExpireAfter: 60,
-		Data:        json.RawMessage([]byte(dataString)),
+		Channel:  "some-channel",
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
 	})
 	s.c.Assert(err, IsNil)
 	reader := bytes.NewReader(packedMessage)

=== modified file 'server/broker/simple/simple_test.go'
--- server/broker/simple/simple_test.go	2014-02-10 23:19:08 +0000
+++ server/broker/simple/simple_test.go	2014-02-10 23:46:49 +0000
@@ -19,6 +19,7 @@
 import (
 	"encoding/json"
 	stdtesting "testing"
+	"time"
 
 	. "launchpad.net/gocheck"
 
@@ -45,8 +46,9 @@
 
 func (s *simpleSuite) TestFeedPending(c *C) {
 	sto := store.NewInMemoryPendingStore()
+	muchLater := time.Now().Add(10 * time.Minute)
 	notification1 := json.RawMessage(`{"m": "M"}`)
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
 	b := NewSimpleBroker(sto, testBrokerConfig, nil)
 	sess := &simpleBrokerSession{
 		exchanges: make(chan broker.Exchange, 1),
@@ -63,8 +65,9 @@
 
 func (s *simpleSuite) TestFeedPendingNop(c *C) {
 	sto := store.NewInMemoryPendingStore()
+	muchLater := time.Now().Add(10 * time.Minute)
 	notification1 := json.RawMessage(`{"m": "M"}`)
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
 	b := NewSimpleBroker(sto, testBrokerConfig, nil)
 	sess := &simpleBrokerSession{
 		exchanges: make(chan broker.Exchange, 1),

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-02-10 23:19:08 +0000
+++ server/broker/testsuite/suite.go	2014-02-10 23:46:49 +0000
@@ -108,7 +108,8 @@
 func (s *CommonBrokerSuite) TestRegistrationFeedPending(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	notification1 := json.RawMessage(`{"m": "M"}`)
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
@@ -156,7 +157,8 @@
 	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-2"})
 	c.Assert(err, IsNil)
 	// add notification to channel *after* the registrations
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
 	b.Broadcast(store.SystemInternalChannelId)
 	select {
 	case <-time.After(5 * time.Second):

=== modified file 'server/store/inmemory.go'
--- server/store/inmemory.go	2014-01-14 15:35:20 +0000
+++ server/store/inmemory.go	2014-02-10 23:46:49 +0000
@@ -19,18 +19,31 @@
 import (
 	"encoding/json"
 	"sync"
+	"time"
 )
 
+// one stored notification
+type notification struct {
+	payload    json.RawMessage
+	expiration time.Time
+}
+
+// one stored channel
+type channel struct {
+	topLevel      int64
+	notifications []notification
+}
+
 // InMemoryPendingStore is a basic in-memory pending notification store.
 type InMemoryPendingStore struct {
 	lock  sync.Mutex
-	store map[InternalChannelId][]json.RawMessage
+	store map[InternalChannelId]*channel
 }
 
 // NewInMemoryPendingStore returns a new InMemoryStore.
 func NewInMemoryPendingStore() *InMemoryPendingStore {
 	return &InMemoryPendingStore{
-		store: make(map[InternalChannelId][]json.RawMessage),
+		store: make(map[InternalChannelId]*channel),
 	}
 }
 
@@ -41,25 +54,40 @@
 	return InternalChannelId(""), ErrUnknownChannel
 }
 
-func (sto *InMemoryPendingStore) AppendToChannel(chanId InternalChannelId, notification json.RawMessage) error {
+func (sto *InMemoryPendingStore) AppendToChannel(chanId InternalChannelId, notificationPayload json.RawMessage, expiration time.Time) error {
 	sto.lock.Lock()
 	defer sto.lock.Unlock()
 	prev := sto.store[chanId]
-	sto.store[chanId] = append(prev, notification)
+	if prev == nil {
+		prev = &channel{}
+	}
+	prev.topLevel++
+	prev.notifications = append(prev.notifications, notification{
+		payload:    notificationPayload,
+		expiration: expiration,
+	})
+	sto.store[chanId] = prev
 	return nil
 }
 
 func (sto *InMemoryPendingStore) GetChannelSnapshot(chanId InternalChannelId) (int64, []json.RawMessage, error) {
 	sto.lock.Lock()
 	defer sto.lock.Unlock()
-	notifications, ok := sto.store[chanId]
+	channel, ok := sto.store[chanId]
 	if !ok {
 		return 0, nil, nil
 	}
-	n := len(notifications)
-	res := make([]json.RawMessage, n)
-	copy(res, notifications)
-	return int64(n), res, nil
+	topLevel := channel.topLevel
+	n := len(channel.notifications)
+	res := make([]json.RawMessage, 0, n)
+	now := time.Now()
+	for _, notification := range channel.notifications {
+		if notification.expiration.Before(now) {
+			continue
+		}
+		res = append(res, notification.payload)
+	}
+	return topLevel, res, nil
 }
 
 // sanity check we implement the interface

=== modified file 'server/store/inmemory_test.go'
--- server/store/inmemory_test.go	2014-02-10 23:19:08 +0000
+++ server/store/inmemory_test.go	2014-02-10 23:46:49 +0000
@@ -18,7 +18,7 @@
 
 import (
 	"encoding/json"
-	// "fmt"
+	"time"
 
 	. "launchpad.net/gocheck"
 )
@@ -54,10 +54,32 @@
 	notification1 := json.RawMessage(`{"a":1}`)
 	notification2 := json.RawMessage(`{"a":2}`)
 
-	sto.AppendToChannel(SystemInternalChannelId, notification1)
-	sto.AppendToChannel(SystemInternalChannelId, notification2)
+	muchLater := time.Now().Add(time.Minute)
+
+	sto.AppendToChannel(SystemInternalChannelId, notification1, muchLater)
+	sto.AppendToChannel(SystemInternalChannelId, notification2, muchLater)
 	top, res, err := sto.GetChannelSnapshot(SystemInternalChannelId)
 	c.Assert(err, IsNil)
 	c.Check(top, Equals, int64(2))
 	c.Check(res, DeepEquals, []json.RawMessage{notification1, notification2})
 }
+
+func (s *inMemorySuite) TestAppendToChannelAndGetChannelSnapshortWithExpiration(c *C) {
+	sto := NewInMemoryPendingStore()
+
+	notification1 := json.RawMessage(`{"a":1}`)
+	notification2 := json.RawMessage(`{"a":2}`)
+
+	verySoon := time.Now().Add(100 * time.Millisecond)
+	muchLater := time.Now().Add(time.Minute)
+
+	sto.AppendToChannel(SystemInternalChannelId, notification1, muchLater)
+	sto.AppendToChannel(SystemInternalChannelId, notification2, verySoon)
+
+	time.Sleep(200 * time.Millisecond)
+
+	top, res, err := sto.GetChannelSnapshot(SystemInternalChannelId)
+	c.Assert(err, IsNil)
+	c.Check(top, Equals, int64(2))
+	c.Check(res, DeepEquals, []json.RawMessage{notification1})
+}

=== modified file 'server/store/store.go'
--- server/store/store.go	2014-01-14 15:35:20 +0000
+++ server/store/store.go	2014-02-10 23:46:49 +0000
@@ -21,6 +21,7 @@
 	"encoding/hex"
 	"encoding/json"
 	"errors"
+	"time"
 )
 
 type InternalChannelId string
@@ -65,8 +66,8 @@
 	// GetInternalChannelId returns the internal store id for a channel
 	// given the name.
 	GetInternalChannelId(name string) (InternalChannelId, error)
-	// AppendToChannel appends a notification to the channel. xxx expiration
-	AppendToChannel(chanId InternalChannelId, notification json.RawMessage) error
+	// AppendToChannel appends a notification to the channel.
+	AppendToChannel(chanId InternalChannelId, notification json.RawMessage, expiration time.Time) error
 	// GetChannelSnapshot gets all the current notifications and
 	// current top level in the channel.
 	GetChannelSnapshot(chanId InternalChannelId) (topLevel int64, payloads []json.RawMessage, err error)