feat(push): persist FCM messages

Signed-off-by: Chris Gianelloni <wolf31o2@blinklabs.io>

Author: wolf31o2

output/push/fcm/message.go

@@ -22,9 +22,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"os"
-
-	"github.com/blinklabs-io/adder/internal/logging"
 )
 
 type Message struct {
@@ -59,10 +56,9 @@ func WithNotification(title string, body string) MessageOption {
 	}
 }
 
-func NewMessage(token string, opts ...MessageOption) *Message {
+func NewMessage(token string, opts ...MessageOption) (*Message, error) {
 	if token == "" {
-		logging.GetLogger().Error("Token is mandatory for FCM message")
-		os.Exit(1)
+		return nil, errors.New("token is mandatory for FCM message")
 	}
 
 	msg := &Message{
@@ -73,7 +69,7 @@ func NewMessage(token string, opts ...MessageOption) *Message {
 	for _, opt := range opts {
 		opt(&msg.MessageContent)
 	}
-	return msg
+	return msg, nil
 }
 
 func Send(accessToken string, projectId string, msg *Message) error {

output/push/fcm/message_test.go

@@ -0,0 +1,64 @@
+// Copyright 2025 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fcm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewMessage(t *testing.T) {
+	t.Run("empty token returns error", func(t *testing.T) {
+		msg, err := NewMessage("")
+		assert.Nil(t, msg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "token is mandatory")
+	})
+
+	t.Run("valid token returns message", func(t *testing.T) {
+		msg, err := NewMessage("valid-token-123")
+		require.NoError(t, err)
+		require.NotNil(t, msg)
+		assert.Equal(t, "valid-token-123", msg.Token)
+	})
+
+	t.Run("with notification option", func(t *testing.T) {
+		msg, err := NewMessage(
+			"valid-token-123",
+			WithNotification("Test Title", "Test Body"),
+		)
+		require.NoError(t, err)
+		require.NotNil(t, msg)
+		require.NotNil(t, msg.Notification)
+		assert.Equal(t, "Test Title", msg.Notification.Title)
+		assert.Equal(t, "Test Body", msg.Notification.Body)
+	})
+
+	t.Run("with data option", func(t *testing.T) {
+		data := map[string]any{
+			"key1": "value1",
+			"key2": "value2",
+		}
+		msg, err := NewMessage(
+			"valid-token-123",
+			WithData(data),
+		)
+		require.NoError(t, err)
+		require.NotNil(t, msg)
+		assert.Equal(t, data, msg.Data)
+	})
+}

output/push/fcm_repository.go

@@ -15,14 +15,21 @@
 package push
 
 import (
+	"encoding/json"
 	"net/http"
+	"os"
+	"sync"
 
 	_ "github.com/blinklabs-io/adder/docs"
+	"github.com/blinklabs-io/adder/internal/logging"
 	"github.com/gin-gonic/gin"
 )
 
 type TokenStore struct {
-	FCMTokens map[string]string
+	FCMTokens    map[string]string `json:"fcm_tokens"`
+	filePath     string
+	mu           sync.RWMutex
+	persistMutex sync.Mutex
 }
 
 // TokenRequest represents a request containing an FCM token.
@@ -43,23 +50,116 @@ type ErrorResponse struct {
 	Error string `json:"error"`
 }
 
-// TODO add support for persistence (#335)
 var fcmStore *TokenStore
 
 func init() {
-	fcmStore = newTokenStore()
+	fcmStore = newTokenStore("")
 }
 
-func newTokenStore() *TokenStore {
-	return &TokenStore{
+func newTokenStore(filePath string) *TokenStore {
+	store := &TokenStore{
 		FCMTokens: make(map[string]string),
+		filePath:  filePath,
+	}
+	// Load existing tokens if persistence is enabled
+	if filePath != "" {
+		store.loadTokens()
+	}
+	return store
+}
+
+// SetPersistenceFile configures the file path for token persistence
+// If called with a non-empty path, tokens will be loaded from and saved to this file
+func SetPersistenceFile(filePath string) {
+	if fcmStore == nil {
+		fcmStore = newTokenStore(filePath)
+		return
+	}
+	fcmStore.persistMutex.Lock()
+	fcmStore.filePath = filePath
+	fcmStore.persistMutex.Unlock()
+	if filePath != "" {
+		fcmStore.loadTokens()
 	}
 }
 
 func getTokenStore() *TokenStore {
 	return fcmStore
 }
 
+// loadTokens loads tokens from the persistence file
+func (s *TokenStore) loadTokens() {
+	s.persistMutex.Lock()
+	filePath := s.filePath
+	s.persistMutex.Unlock()
+
+	if filePath == "" {
+		return
+	}
+
+	logger := logging.GetLogger()
+
+	data, err := os.ReadFile(filePath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			// File doesn't exist yet, that's fine for first run
+			logger.Debug("FCM token persistence file does not exist yet", "path", filePath)
+			return
+		}
+		logger.Error("failed to read FCM tokens from file", "error", err, "path", filePath)
+		return
+	}
+
+	var loadedStore struct {
+		FCMTokens map[string]string `json:"fcm_tokens"`
+	}
+	if err := json.Unmarshal(data, &loadedStore); err != nil {
+		logger.Error("failed to parse FCM tokens from file", "error", err, "path", filePath)
+		return
+	}
+
+	s.mu.Lock()
+	if loadedStore.FCMTokens != nil {
+		s.FCMTokens = loadedStore.FCMTokens
+	}
+	s.mu.Unlock()
+
+	logger.Info("loaded FCM tokens from persistence file", "count", len(loadedStore.FCMTokens), "path", filePath)
+}
+
+// saveTokens saves tokens to the persistence file
+func (s *TokenStore) saveTokens() {
+	s.persistMutex.Lock()
+	filePath := s.filePath
+	s.persistMutex.Unlock()
+
+	if filePath == "" {
+		return
+	}
+
+	logger := logging.GetLogger()
+
+	s.mu.RLock()
+	data, err := json.MarshalIndent(struct {
+		FCMTokens map[string]string `json:"fcm_tokens"`
+	}{
+		FCMTokens: s.FCMTokens,
+	}, "", "  ")
+	s.mu.RUnlock()
+
+	if err != nil {
+		logger.Error("failed to marshal FCM tokens", "error", err)
+		return
+	}
+
+	if err := os.WriteFile(filePath, data, 0o600); err != nil {
+		logger.Error("failed to write FCM tokens to file", "error", err, "path", filePath)
+		return
+	}
+
+	logger.Debug("saved FCM tokens to persistence file", "path", filePath)
+}
+
 // @Summary		Store FCM Token
 // @Description	Store a new FCM token
 // @Accept			json
@@ -84,7 +184,10 @@ func storeFCMToken(c *gin.Context) {
 		)
 		return
 	}
+	store.mu.Lock()
 	store.FCMTokens[req.FCMToken] = req.FCMToken
+	store.mu.Unlock()
+	store.saveTokens()
 	c.Status(http.StatusCreated)
 }
 
@@ -106,7 +209,9 @@ func readFCMToken(c *gin.Context) {
 		)
 		return
 	}
+	store.mu.RLock()
 	storedToken, exists := store.FCMTokens[token]
+	store.mu.RUnlock()
 	if !exists {
 		c.Status(http.StatusNotFound)
 		return
@@ -132,20 +237,32 @@ func deleteFCMToken(c *gin.Context) {
 		)
 		return
 	}
+	store.mu.Lock()
 	_, exists := store.FCMTokens[token]
 	if exists {
 		delete(store.FCMTokens, token)
+	}
+	store.mu.Unlock()
+	if exists {
+		store.saveTokens()
 		c.Status(http.StatusNoContent)
 	} else {
 		c.Status(http.StatusNotFound)
 	}
 }
 
-// GetFcmTokens returns the current in-memory FCM tokens
+// GetFcmTokens returns a copy of the current in-memory FCM tokens
 func GetFcmTokens() map[string]string {
 	store := getTokenStore()
 	if store == nil {
 		return make(map[string]string)
 	}
-	return store.FCMTokens
+	store.mu.RLock()
+	defer store.mu.RUnlock()
+	// Return a copy to avoid race conditions
+	tokens := make(map[string]string, len(store.FCMTokens))
+	for k, v := range store.FCMTokens {
+		tokens[k] = v
+	}
+	return tokens
 }

output/push/plugin.go

@@ -52,12 +52,19 @@ func init() {
 }
 
 func NewFromCmdlineOptions() plugin.Plugin {
-	p := New(
+	p, err := New(
 		WithLogger(
 			logging.GetLogger().With("plugin", "output.push"),
 		),
 		WithAccessTokenUrl(cmdlineOptions.accessTokenUrl),
 		WithServiceAccountFilePath(cmdlineOptions.serviceAccountFilePath),
 	)
+	if err != nil {
+		logging.GetLogger().Error(
+			"failed to create push output plugin",
+			"error", err,
+		)
+		return nil
+	}
 	return p
 }