Subscriber Billed Topics

server/errors.go

@@ -92,4 +92,5 @@ var (
 	errHTTPInternalError                             = &errHTTP{50001, http.StatusInternalServerError, "internal server error", ""}
 	errHTTPInternalErrorInvalidPath                  = &errHTTP{50002, http.StatusInternalServerError, "internal server error: invalid path", ""}
 	errHTTPInternalErrorMissingBaseURL               = &errHTTP{50003, http.StatusInternalServerError, "internal server error: base-url must be be configured for this feature", "https://ntfy.sh/docs/config/"}
+	errHTTPWontStoreMessage                          = &errHTTP{50701, http.StatusInsufficientStorage, "topic is inactive; no device available to recieve message", ""}
 )

server/server.go

@@ -114,13 +114,15 @@ var (
 )
 
 const (
-	firebaseControlTopic     = "~control"                // See Android if changed
-	firebasePollTopic        = "~poll"                   // See iOS if changed
-	emptyMessageBody         = "triggered"               // Used if message body is empty
-	newMessageBody           = "New message"             // Used in poll requests as generic message
-	defaultAttachmentMessage = "You received a file: %s" // Used if message body is empty, and there is an attachment
-	encodingBase64           = "base64"                  // Used mainly for binary UnifiedPush messages
-	jsonBodyBytesLimit       = 16384
+	firebaseControlTopic        = "~control"                // See Android if changed
+	firebasePollTopic           = "~poll"                   // See iOS if changed
+	emptyMessageBody            = "triggered"               // Used if message body is empty
+	newMessageBody              = "New message"             // Used in poll requests as generic message
+	defaultAttachmentMessage    = "You received a file: %s" // Used if message body is empty, and there is an attachment
+	encodingBase64              = "base64"                  // Used mainly for binary UnifiedPush messages
+	jsonBodyBytesLimit          = 16384
+	subscriberBilledTopicPrefix = "up_"
+	subscriberBilledValidity    = 12 * time.Hour
 )
 
 // WebSocket constants
@@ -370,6 +372,7 @@ func (s *Server) handleError(w http.ResponseWriter, r *http.Request, v *visitor,
 	}
 	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
+	w.Header().Set("TTL", "0")                                                       // if message is not being stored because of an error, tell them
 	w.WriteHeader(httpErr.HTTPCode)
 	io.WriteString(w, httpErr.JSON()+"\n")
 }
@@ -603,6 +606,14 @@ func (s *Server) handlePublishWithoutResponse(r *http.Request, v *visitor) (*mes
 	if err != nil {
 		return nil, err
 	}
+	v_old := v
+	if strings.HasPrefix(t.ID, subscriberBilledTopicPrefix) {
+		v = t.getBillee()
+		if v == nil {
+			return nil, errHTTPWontStoreMessage
+		}
+	}
+
 	if !v.MessageAllowed() {
 		return nil, errHTTPTooManyRequestsLimitMessages
 	}
@@ -620,7 +631,7 @@ func (s *Server) handlePublishWithoutResponse(r *http.Request, v *visitor) (*mes
 	}
 	m.Sender = v.IP()
 	m.User = v.MaybeUserID()
-	m.Expires = time.Now().Add(v.Limits().MessageExpiryDuration).Unix()
+	m.Expires = time.Unix(m.Time, 0).Add(v.Limits().MessageExpiryDuration).Unix()
 	if err := s.handlePublishBody(r, v, m, body, unifiedpush); err != nil {
 		return nil, err
 	}
@@ -637,8 +648,9 @@ func (s *Server) handlePublishWithoutResponse(r *http.Request, v *visitor) (*mes
 			"message_email":       email,
 		}).
 		Debug("Received message")
+		//Where should I log the original visitor vs the billing visitor
 	if log.IsTrace() {
-		logvrm(v, r, m).
+		logvrm(v_old, r, m).
 			Tag(tagPublish).
 			Field("message_body", util.MaybeMarshalJSON(m)).
 			Trace("Message body")
@@ -664,6 +676,8 @@ func (s *Server) handlePublishWithoutResponse(r *http.Request, v *visitor) (*mes
 		if err := s.messageCache.AddMessage(m); err != nil {
 			return nil, err
 		}
+	} else {
+		m.Expires = m.Time
 	}
 	u := v.User()
 	if s.userManager != nil && u != nil && u.Tier != nil {
@@ -680,6 +694,10 @@ func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visito
 	if err != nil {
 		return err
 	}
+
+	w.Header().Set("TTL", strconv.FormatInt(m.Expires-m.Time, 10)) // return how long a message will be stored for
+
+	// using m.Time, not time.Now() so the value isn't negative if the request is processed at a second boundary
 	return s.writeJSON(w, m)
 }
 
@@ -1023,7 +1041,7 @@ func (s *Server) handleSubscribeHTTP(w http.ResponseWriter, r *http.Request, v *
 	defer cancel()
 	subscriberIDs := make([]int, 0)
 	for _, t := range topics {
-		subscriberIDs = append(subscriberIDs, t.Subscribe(sub, v.MaybeUserID(), cancel))
+		subscriberIDs = append(subscriberIDs, t.Subscribe(sub, v, cancel))
 	}
 	defer func() {
 		for i, subscriberID := range subscriberIDs {
@@ -1155,7 +1173,7 @@ func (s *Server) handleSubscribeWS(w http.ResponseWriter, r *http.Request, v *vi
 	}
 	subscriberIDs := make([]int, 0)
 	for _, t := range topics {
-		subscriberIDs = append(subscriberIDs, t.Subscribe(sub, v.MaybeUserID(), cancel))
+		subscriberIDs = append(subscriberIDs, t.Subscribe(sub, v, cancel))
 	}
 	defer func() {
 		for i, subscriberID := range subscriberIDs {
@@ -1402,9 +1420,10 @@ func (s *Server) execManager() {
 			defer s.mu.Unlock()
 			for _, t := range s.topics {
 				subs := t.SubscribersCount()
-				log.Tag(tagManager).Trace("- topic %s: %d subscribers", t.ID, subs)
+				expiryTime := time.Until(t.lastVisitorExpires)
+				log.Tag(tagManager).Trace("- topic %s: %d subscribers, expires in %s", t.ID, subs, expiryTime)
 				msgs, exists := messageCounts[t.ID]
-				if subs == 0 && (!exists || msgs == 0) {
+				if t.Stale() && (!exists || msgs == 0) {
 					log.Tag(tagManager).Trace("Deleting empty topic %s", t.ID)
 					emptyTopics++
 					delete(s.topics, t.ID)

server/topic.go

@@ -1,22 +1,26 @@
 package server
 
 import (
-	"heckel.io/ntfy/log"
 	"math/rand"
 	"sync"
+	"time"
+
+	"heckel.io/ntfy/log"
 )
 
 // topic represents a channel to which subscribers can subscribe, and publishers
 // can publish a message
 type topic struct {
-	ID          string
-	subscribers map[int]*topicSubscriber
-	mu          sync.Mutex
+	ID                 string
+	subscribers        map[int]*topicSubscriber
+	lastVisitor        *visitor
+	lastVisitorExpires time.Time
+	mu                 sync.Mutex
 }
 
 type topicSubscriber struct {
-	userID     string // User ID associated with this subscription, may be empty
 	subscriber subscriber
+	visitor    *visitor // User ID associated with this subscription, may be empty
 	cancel     func()
 }
 
@@ -32,22 +36,42 @@ func newTopic(id string) *topic {
 }
 
 // Subscribe subscribes to this topic
-func (t *topic) Subscribe(s subscriber, userID string, cancel func()) int {
+func (t *topic) Subscribe(s subscriber, visitor *visitor, cancel func()) int {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	subscriberID := rand.Int()
 	t.subscribers[subscriberID] = &topicSubscriber{
-		userID:     userID, // May be empty
+		visitor:    visitor, // May be empty
 		subscriber: s,
 		cancel:     cancel,
 	}
 	return subscriberID
 }
 
+func (t *topic) Stale() bool {
+	return t.getBillee() == nil
+}
+
+func (t *topic) getBillee() *visitor {
+	for _, this_subscriber := range t.subscribers {
+		return this_subscriber.visitor
+	}
+	if t.lastVisitor != nil && t.lastVisitorExpires.After(time.Now()) {
+		t.lastVisitor = nil
+	}
+	return t.lastVisitor
+
+}
+
 // Unsubscribe removes the subscription from the list of subscribers
 func (t *topic) Unsubscribe(id int) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
+
+	if len(t.subscribers) == 1 {
+		t.lastVisitor = t.subscribers[id].visitor
+		t.lastVisitorExpires = time.Now().Add(subscriberBilledValidity)
+	}
 	delete(t.subscribers, id)
 }
 
@@ -87,8 +111,9 @@ func (t *topic) CancelSubscribers(exceptUserID string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	for _, s := range t.subscribers {
-		if s.userID != exceptUserID {
-			log.Tag(tagSubscribe).Field("topic", t.ID).Debug("Canceling subscriber %s", s.userID)
+		if s.visitor.MaybeUserID() != exceptUserID {
+			// TODO: Shouldn't this log the IP for anonymous visitors? It was s.userID before my change.
+			log.Tag(tagSubscribe).Field("topic", t.ID).Debug("Canceling subscriber %s", s.visitor.MaybeUserID())
 			s.cancel()
 		}
 	}
@@ -101,7 +126,7 @@ func (t *topic) subscribersCopy() map[int]*topicSubscriber {
 	subscribers := make(map[int]*topicSubscriber)
 	for k, sub := range t.subscribers {
 		subscribers[k] = &topicSubscriber{
-			userID:     sub.userID,
+			visitor:    sub.visitor,
 			subscriber: sub.subscriber,
 			cancel:     sub.cancel,
 		}