Notifications repository + service (#4565)

* Notifications repository

* feat: notifications repo + service

* fix: remove use of notifications/notifications.go

* Update persistence/notificationsrepository/notificationsrepository.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* fix: remove unused method

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

core/core.go

@@ -15,8 +15,8 @@ import (
 	"github.com/owncast/owncast/core/transcoder"
 	"github.com/owncast/owncast/core/webhooks"
 	"github.com/owncast/owncast/models"
-	"github.com/owncast/owncast/notifications"
 	"github.com/owncast/owncast/persistence/configrepository"
+	"github.com/owncast/owncast/persistence/notificationsrepository"
 	"github.com/owncast/owncast/persistence/tables"
 	"github.com/owncast/owncast/utils"
 	"github.com/owncast/owncast/yp"
@@ -83,7 +83,7 @@ func Start() error {
 
 	webhooks.SetupWebhooks(GetStatus)
 
-	notifications.Setup(data.GetStore())
+	notificationsrepository.Setup(data.GetStore())
 
 	return nil
 }

core/streamState.go

@@ -15,8 +15,8 @@ import (
 	"github.com/owncast/owncast/core/transcoder"
 	"github.com/owncast/owncast/core/webhooks"
 	"github.com/owncast/owncast/models"
-	"github.com/owncast/owncast/notifications"
 	"github.com/owncast/owncast/persistence/configrepository"
+	"github.com/owncast/owncast/services/notifications"
 	"github.com/owncast/owncast/utils"
 )
 
@@ -189,10 +189,10 @@ func startLiveStreamNotificationsTimer() context.CancelFunc {
 			}
 
 			// Send notification to those who have registered for them.
-			if notifier, err := notifications.New(data.GetDatastore()); err != nil {
+			if notificationService, err := notifications.New(data.GetDatastore()); err != nil {
 				log.Errorln(err)
 			} else {
-				notifier.Notify()
+				notificationService.Notify()
 			}
 
 			now := time.Now()

notifications/notifications.go

@@ -1,190 +0,0 @@
-package notifications
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/owncast/owncast/config"
-	"github.com/owncast/owncast/core/data"
-	"github.com/owncast/owncast/db"
-	"github.com/owncast/owncast/models"
-	"github.com/owncast/owncast/persistence/configrepository"
-	"github.com/owncast/owncast/persistence/tables"
-
-	"github.com/owncast/owncast/notifications/browser"
-	"github.com/owncast/owncast/notifications/discord"
-	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
-)
-
-// Notifier is an instance of the live stream notifier.
-type Notifier struct {
-	datastore        *data.Datastore
-	browser          *browser.Browser
-	discord          *discord.Discord
-	configRepository configrepository.ConfigRepository
-}
-
-// Setup will perform any pre-use setup for the notifier.
-func Setup(datastore *data.Datastore) {
-	tables.CreateNotificationsTable(datastore.DB)
-	initializeBrowserPushIfNeeded()
-}
-
-func initializeBrowserPushIfNeeded() {
-	configRepository := configrepository.Get()
-
-	pubKey, _ := configRepository.GetBrowserPushPublicKey()
-	privKey, _ := configRepository.GetBrowserPushPrivateKey()
-
-	// We need browser push keys so people can register for pushes.
-	if pubKey == "" || privKey == "" {
-		browserPrivateKey, browserPublicKey, err := browser.GenerateBrowserPushKeys()
-		if err != nil {
-			log.Errorln("unable to initialize browser push notification keys", err)
-		}
-
-		if err := configRepository.SetBrowserPushPrivateKey(browserPrivateKey); err != nil {
-			log.Errorln("unable to set browser push private key", err)
-		}
-
-		if err := configRepository.SetBrowserPushPublicKey(browserPublicKey); err != nil {
-			log.Errorln("unable to set browser push public key", err)
-		}
-	}
-
-	// Enable browser push notifications by default.
-	if !configRepository.GetHasPerformedInitialNotificationsConfig() {
-		_ = configRepository.SetBrowserPushConfig(models.BrowserNotificationConfiguration{Enabled: true, GoLiveMessage: config.GetDefaults().FederationGoLiveMessage})
-		_ = configRepository.SetHasPerformedInitialNotificationsConfig(true)
-	}
-}
-
-// New creates a new instance of the Notifier.
-func New(datastore *data.Datastore) (*Notifier, error) {
-	notifier := Notifier{
-		datastore:        datastore,
-		configRepository: configrepository.Get(),
-	}
-
-	if err := notifier.setupBrowserPush(); err != nil {
-		log.Error(err)
-	}
-	if err := notifier.setupDiscord(); err != nil {
-		log.Error(err)
-	}
-
-	return &notifier, nil
-}
-
-func (n *Notifier) setupBrowserPush() error {
-	if n.configRepository.GetBrowserPushConfig().Enabled {
-		publicKey, err := n.configRepository.GetBrowserPushPublicKey()
-		if err != nil || publicKey == "" {
-			return errors.Wrap(err, "browser notifier disabled, failed to get browser push public key")
-		}
-
-		privateKey, err := n.configRepository.GetBrowserPushPrivateKey()
-		if err != nil || privateKey == "" {
-			return errors.Wrap(err, "browser notifier disabled, failed to get browser push private key")
-		}
-
-		browserNotifier, err := browser.New(n.datastore, publicKey, privateKey)
-		if err != nil {
-			return errors.Wrap(err, "error creating browser notifier")
-		}
-		n.browser = browserNotifier
-	}
-	return nil
-}
-
-func (n *Notifier) notifyBrowserPush() {
-	destinations, err := GetNotificationDestinationsForChannel(BrowserPushNotification)
-	if err != nil {
-		log.Errorln("error getting browser push notification destinations", err)
-	}
-	for _, destination := range destinations {
-		unsubscribed, err := n.browser.Send(destination, n.configRepository.GetServerName(), n.configRepository.GetBrowserPushConfig().GoLiveMessage)
-		if unsubscribed {
-			// If the error is "unsubscribed", then remove the destination from the database.
-			if err := RemoveNotificationForChannel(BrowserPushNotification, destination); err != nil {
-				log.Errorln(err)
-			}
-		} else if err != nil {
-			log.Errorln(err)
-		}
-	}
-}
-
-func (n *Notifier) setupDiscord() error {
-	discordConfig := n.configRepository.GetDiscordConfig()
-	if discordConfig.Enabled && discordConfig.Webhook != "" {
-		var image string
-		if serverURL := n.configRepository.GetServerURL(); serverURL != "" {
-			image = serverURL + "/logo"
-		}
-		discordNotifier, err := discord.New(
-			n.configRepository.GetServerName(),
-			image,
-			discordConfig.Webhook,
-		)
-		if err != nil {
-			return errors.Wrap(err, "error creating discord notifier")
-		}
-		n.discord = discordNotifier
-	}
-	return nil
-}
-
-func (n *Notifier) notifyDiscord() {
-	goLiveMessage := n.configRepository.GetDiscordConfig().GoLiveMessage
-	streamTitle := n.configRepository.GetStreamTitle()
-	if streamTitle != "" {
-		goLiveMessage += "\n" + streamTitle
-	}
-	message := fmt.Sprintf("%s\n\n%s", goLiveMessage, n.configRepository.GetServerURL())
-
-	if err := n.discord.Send(message); err != nil {
-		log.Errorln("error sending discord message", err)
-	}
-}
-
-// Notify will fire the different notification channels.
-func (n *Notifier) Notify() {
-	if n.browser != nil {
-		n.notifyBrowserPush()
-	}
-
-	if n.discord != nil {
-		n.notifyDiscord()
-	}
-}
-
-// RemoveNotificationForChannel removes a notification destination.
-func RemoveNotificationForChannel(channel, destination string) error {
-	log.Debugln("Removing notification for channel", channel)
-
-	return data.GetDatastore().GetQueries().RemoveNotificationDestinationForChannel(context.Background(), db.RemoveNotificationDestinationForChannelParams{
-		Channel:     channel,
-		Destination: destination,
-	})
-}
-
-// GetNotificationDestinationsForChannel will return a collection of
-// destinations to notify for a given channel.
-func GetNotificationDestinationsForChannel(channel string) ([]string, error) {
-	result, err := data.GetDatastore().GetQueries().GetNotificationDestinationsForChannel(context.Background(), channel)
-	if err != nil {
-		return nil, errors.Wrap(err, "unable to query notification destinations for channel "+channel)
-	}
-
-	return result, nil
-}
-
-// AddNotification saves a new user notification destination.
-func AddNotification(channel, destination string) error {
-	return data.GetDatastore().GetQueries().AddNotification(context.Background(), db.AddNotificationParams{
-		Channel:     channel,
-		Destination: destination,
-	})
-}

persistence/notificationsrepository/channels.go

@@ -1,4 +1,4 @@
-package notifications
+package notificationsrepository
 
 const (
 	// BrowserPushNotification represents a push notification for a browser.

persistence/notificationsrepository/integration_test.go

@@ -0,0 +1,231 @@
+package notificationsrepository
+
+import (
+	"testing"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/persistence/configrepository"
+)
+
+var (
+	integrationTestDatastore *data.Datastore
+	integrationRepo          NotificationsRepository
+)
+
+func init() {
+	// Use the shared test datastore from the main test file
+	// This ensures we're using the same database setup
+	integrationTestDatastore = testDatastore
+	if integrationTestDatastore != nil {
+		integrationRepo = New(integrationTestDatastore)
+	}
+}
+
+func TestIntegrationSetup(t *testing.T) {
+	// This test ensures integration setup works properly
+	// The actual setup is done in the init function below
+	if integrationTestDatastore == nil {
+		integrationTestDatastore = data.GetDatastore()
+		Setup(integrationTestDatastore)
+		integrationRepo = New(integrationTestDatastore)
+	}
+
+	if integrationRepo == nil {
+		t.Error("Integration repository should be initialized")
+	}
+}
+
+func TestBrowserPushSetupIntegration(t *testing.T) {
+	// Test that browser push keys are generated during setup
+	configRepo := configrepository.Get()
+	pubKey, err := configRepo.GetBrowserPushPublicKey()
+	if err != nil {
+		t.Errorf("Should be able to get browser push public key: %v", err)
+	}
+
+	privKey, err := configRepo.GetBrowserPushPrivateKey()
+	if err != nil {
+		t.Errorf("Should be able to get browser push private key: %v", err)
+	}
+
+	// Keys should be generated during setup
+	if pubKey == "" || privKey == "" {
+		t.Error("Browser push keys should be generated during setup")
+	}
+}
+
+func TestBrowserPushConfigurationIntegration(t *testing.T) {
+	configRepo := configrepository.Get()
+
+	// Test that browser push is enabled by default
+	browserConfig := configRepo.GetBrowserPushConfig()
+	if !browserConfig.Enabled {
+		t.Error("Browser push should be enabled by default")
+	}
+
+	// Test that initial notification configuration flag is set
+	hasConfigured := configRepo.GetHasPerformedInitialNotificationsConfig()
+	if !hasConfigured {
+		t.Error("Should have performed initial notifications configuration")
+	}
+}
+
+func TestDiscordConfigurationIntegration(t *testing.T) {
+	configRepo := configrepository.Get()
+
+	// Test Discord configuration setup
+	discordConfig := models.DiscordConfiguration{
+		Enabled:       true,
+		Webhook:       "https://discord.com/api/webhooks/test",
+		GoLiveMessage: "Test stream is live!",
+	}
+
+	err := configRepo.SetDiscordConfig(discordConfig)
+	if err != nil {
+		t.Errorf("Failed to set Discord configuration: %v", err)
+	}
+
+	// Test with disabled Discord
+	disabledConfig := models.DiscordConfiguration{
+		Enabled: false,
+		Webhook: "",
+	}
+	err = configRepo.SetDiscordConfig(disabledConfig)
+	if err != nil {
+		t.Errorf("Failed to disable Discord configuration: %v", err)
+	}
+}
+
+func TestNotificationWorkflowIntegration(t *testing.T) {
+	channel := BrowserPushNotification
+	destination := "integration-test-endpoint"
+
+	// Add a notification
+	err := integrationRepo.AddNotification(channel, destination)
+	if err != nil {
+		t.Errorf("Failed to add notification: %v", err)
+	}
+
+	// Verify it was added
+	destinations, err := integrationRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations: %v", err)
+	}
+
+	found := false
+	for _, dest := range destinations {
+		if dest == destination {
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		t.Error("Notification destination should be found after adding")
+	}
+
+	// Clean up
+	err = integrationRepo.RemoveNotificationForChannel(channel, destination)
+	if err != nil {
+		t.Errorf("Failed to remove notification: %v", err)
+	}
+
+	// Verify removal
+	destinationsAfter, err := integrationRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations after removal: %v", err)
+	}
+
+	for _, dest := range destinationsAfter {
+		if dest == destination {
+			t.Error("Notification destination should be removed")
+		}
+	}
+}
+
+func TestRepositoryGetSingleton(t *testing.T) {
+	// Test that Get() returns the same instance
+	repo1 := Get()
+	repo2 := Get()
+
+	if repo1 != repo2 {
+		t.Error("Get() should return the same singleton instance")
+	}
+
+	// Test that it implements the interface
+	var _ NotificationsRepository = repo1
+}
+
+func TestDatabaseTransactionIntegrity(t *testing.T) {
+	channel := "TRANSACTION_TEST_CHANNEL"
+	destinations := []string{"dest1", "dest2", "dest3"}
+
+	// Add multiple notifications
+	for _, dest := range destinations {
+		err := integrationRepo.AddNotification(channel, dest)
+		if err != nil {
+			t.Errorf("Failed to add notification %s: %v", dest, err)
+		}
+	}
+
+	// Verify all were added
+	retrievedDests, err := integrationRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to retrieve destinations: %v", err)
+	}
+
+	if len(retrievedDests) != len(destinations) {
+		t.Errorf("Expected %d destinations, got %d", len(destinations), len(retrievedDests))
+	}
+
+	// Remove them one by one and verify state consistency
+	for i, dest := range destinations {
+		err := integrationRepo.RemoveNotificationForChannel(channel, dest)
+		if err != nil {
+			t.Errorf("Failed to remove notification %s: %v", dest, err)
+		}
+
+		remaining, err := integrationRepo.GetNotificationDestinationsForChannel(channel)
+		if err != nil {
+			t.Errorf("Failed to get remaining destinations: %v", err)
+		}
+
+		expectedRemaining := len(destinations) - (i + 1)
+		if len(remaining) != expectedRemaining {
+			t.Errorf("Expected %d remaining destinations, got %d", expectedRemaining, len(remaining))
+		}
+	}
+}
+
+func TestErrorHandling(t *testing.T) {
+	// Test with empty channel
+	err := integrationRepo.AddNotification("", "test-destination")
+	if err != nil {
+		// Empty channel might be allowed - this documents the behavior
+		t.Logf("Empty channel behavior: %v", err)
+	}
+
+	// Test with empty destination
+	err = integrationRepo.AddNotification("TEST_CHANNEL", "")
+	if err != nil {
+		// Empty destination might be allowed - this documents the behavior
+		t.Logf("Empty destination behavior: %v", err)
+	}
+
+	// Test removing from empty channel
+	err = integrationRepo.RemoveNotificationForChannel("", "test-destination")
+	if err != nil {
+		t.Logf("Remove from empty channel behavior: %v", err)
+	}
+
+	// Test getting destinations for empty channel
+	destinations, err := integrationRepo.GetNotificationDestinationsForChannel("")
+	if err != nil {
+		t.Errorf("Getting destinations for empty channel should not error: %v", err)
+	}
+
+	if len(destinations) > 0 {
+		t.Logf("Empty channel returned %d destinations", len(destinations))
+	}
+}

persistence/notificationsrepository/notificationsrepository.go

@@ -0,0 +1,110 @@
+package notificationsrepository
+
+import (
+	"context"
+
+	"github.com/owncast/owncast/config"
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/persistence/configrepository"
+	"github.com/owncast/owncast/persistence/tables"
+
+	"github.com/owncast/owncast/notifications/browser"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+type NotificationsRepository interface {
+	AddNotification(channel, destination string) error
+	RemoveNotificationForChannel(channel, destination string) error
+	GetNotificationDestinationsForChannel(channel string) ([]string, error)
+}
+
+// SqlNotificationsRepository handles database operations for notifications.
+type SqlNotificationsRepository struct {
+	datastore *data.Datastore
+}
+
+// NOTE: This is temporary during the transition period.
+var temporaryGlobalInstance NotificationsRepository
+
+// Get will return the notifications repository.
+func Get() NotificationsRepository {
+	if temporaryGlobalInstance == nil {
+		i := New(data.GetDatastore())
+		temporaryGlobalInstance = i
+	}
+	return temporaryGlobalInstance
+}
+
+// Setup will perform any pre-use setup for the notifier.
+func Setup(datastore *data.Datastore) {
+	tables.CreateNotificationsTable(datastore.DB)
+	initializeBrowserPushIfNeeded()
+}
+
+func initializeBrowserPushIfNeeded() {
+	configRepository := configrepository.Get()
+
+	pubKey, _ := configRepository.GetBrowserPushPublicKey()
+	privKey, _ := configRepository.GetBrowserPushPrivateKey()
+
+	// We need browser push keys so people can register for pushes.
+	if pubKey == "" || privKey == "" {
+		browserPrivateKey, browserPublicKey, err := browser.GenerateBrowserPushKeys()
+		if err != nil {
+			log.Errorln("unable to initialize browser push notification keys", err)
+		}
+
+		if err := configRepository.SetBrowserPushPrivateKey(browserPrivateKey); err != nil {
+			log.Errorln("unable to set browser push private key", err)
+		}
+
+		if err := configRepository.SetBrowserPushPublicKey(browserPublicKey); err != nil {
+			log.Errorln("unable to set browser push public key", err)
+		}
+	}
+
+	// Enable browser push notifications by default.
+	if !configRepository.GetHasPerformedInitialNotificationsConfig() {
+		_ = configRepository.SetBrowserPushConfig(models.BrowserNotificationConfiguration{Enabled: true, GoLiveMessage: config.GetDefaults().FederationGoLiveMessage})
+		_ = configRepository.SetHasPerformedInitialNotificationsConfig(true)
+	}
+}
+
+// New creates a new instance of the NotificationsRepository.
+func New(datastore *data.Datastore) NotificationsRepository {
+	return &SqlNotificationsRepository{
+		datastore: datastore,
+	}
+}
+
+// AddNotification saves a new user notification destination.
+func (n *SqlNotificationsRepository) AddNotification(channel, destination string) error {
+	return n.datastore.GetQueries().AddNotification(context.Background(), db.AddNotificationParams{
+		Channel:     channel,
+		Destination: destination,
+	})
+}
+
+// RemoveNotificationForChannel removes a notification destination.
+func (n *SqlNotificationsRepository) RemoveNotificationForChannel(channel, destination string) error {
+	log.Debugln("Removing notification for channel", channel)
+
+	return n.datastore.GetQueries().RemoveNotificationDestinationForChannel(context.Background(), db.RemoveNotificationDestinationForChannelParams{
+		Channel:     channel,
+		Destination: destination,
+	})
+}
+
+// GetNotificationDestinationsForChannel will return a collection of
+// destinations to notify for a given channel.
+func (n *SqlNotificationsRepository) GetNotificationDestinationsForChannel(channel string) ([]string, error) {
+	result, err := n.datastore.GetQueries().GetNotificationDestinationsForChannel(context.Background(), channel)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to query notification destinations for channel "+channel)
+	}
+
+	return result, nil
+}

persistence/notificationsrepository/notificationsrepository_test.go

@@ -0,0 +1,323 @@
+package notificationsrepository
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/owncast/owncast/core/data"
+)
+
+var (
+	testDatastore *data.Datastore
+	testRepo      NotificationsRepository
+)
+
+func TestMain(m *testing.M) {
+	// Create an in-memory database for testing
+	if err := data.SetupPersistence(":memory:"); err != nil {
+		panic(err)
+	}
+
+	// Get the shared datastore instance
+	testDatastore = data.GetDatastore()
+
+	// Setup the notifications repository
+	Setup(testDatastore)
+	testRepo = New(testDatastore)
+
+	// Run tests
+	m.Run()
+}
+
+func TestAddNotification(t *testing.T) {
+	channel := "TEST_CHANNEL_ADD_UNIQUE"
+	destination := "test-destination-add-unique-123"
+
+	// Test adding a notification
+	err := testRepo.AddNotification(channel, destination)
+	if err != nil {
+		t.Errorf("Failed to add notification: %v", err)
+	}
+
+	// Verify the notification was added
+	destinations, err := testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations: %v", err)
+	}
+
+	if len(destinations) != 1 {
+		t.Errorf("Expected 1 destination, got %d", len(destinations))
+	}
+
+	if destinations[0] != destination {
+		t.Errorf("Expected destination %s, got %s", destination, destinations[0])
+	}
+}
+
+func TestAddMultipleNotifications(t *testing.T) {
+	channel := "MULTI_TEST_CHANNEL"
+	destinations := []string{"dest-1", "dest-2", "dest-3"}
+
+	// Add multiple notifications
+	for _, dest := range destinations {
+		err := testRepo.AddNotification(channel, dest)
+		if err != nil {
+			t.Errorf("Failed to add notification for destination %s: %v", dest, err)
+		}
+	}
+
+	// Verify all notifications were added
+	retrievedDests, err := testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations: %v", err)
+	}
+
+	if len(retrievedDests) != len(destinations) {
+		t.Errorf("Expected %d destinations, got %d", len(destinations), len(retrievedDests))
+	}
+
+	// Check that all destinations are present
+	for _, expectedDest := range destinations {
+		found := false
+		for _, retrievedDest := range retrievedDests {
+			if retrievedDest == expectedDest {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Errorf("Expected destination %s not found in retrieved destinations", expectedDest)
+		}
+	}
+}
+
+func TestRemoveNotificationForChannel(t *testing.T) {
+	channel := "REMOVE_TEST_CHANNEL"
+	destination1 := "remove-dest-1"
+	destination2 := "remove-dest-2"
+
+	// Add two notifications
+	err := testRepo.AddNotification(channel, destination1)
+	if err != nil {
+		t.Errorf("Failed to add notification: %v", err)
+	}
+
+	err = testRepo.AddNotification(channel, destination2)
+	if err != nil {
+		t.Errorf("Failed to add notification: %v", err)
+	}
+
+	// Verify both were added
+	destinations, err := testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations: %v", err)
+	}
+
+	if len(destinations) != 2 {
+		t.Errorf("Expected 2 destinations before removal, got %d", len(destinations))
+	}
+
+	// Remove one notification
+	err = testRepo.RemoveNotificationForChannel(channel, destination1)
+	if err != nil {
+		t.Errorf("Failed to remove notification: %v", err)
+	}
+
+	// Verify only one remains
+	destinations, err = testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations after removal: %v", err)
+	}
+
+	if len(destinations) != 1 {
+		t.Errorf("Expected 1 destination after removal, got %d", len(destinations))
+	}
+
+	if destinations[0] != destination2 {
+		t.Errorf("Expected remaining destination %s, got %s", destination2, destinations[0])
+	}
+}
+
+func TestGetNotificationDestinationsForNonExistentChannel(t *testing.T) {
+	channel := "NON_EXISTENT_CHANNEL"
+
+	destinations, err := testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations for non-existent channel: %v", err)
+	}
+
+	if len(destinations) != 0 {
+		t.Errorf("Expected 0 destinations for non-existent channel, got %d", len(destinations))
+	}
+}
+
+func TestRemoveNonExistentNotification(t *testing.T) {
+	channel := "NON_EXISTENT_REMOVE_CHANNEL"
+	destination := "non-existent-destination"
+
+	// Try to remove a notification that doesn't exist
+	err := testRepo.RemoveNotificationForChannel(channel, destination)
+	if err != nil {
+		t.Errorf("Removing non-existent notification should not return error, got: %v", err)
+	}
+}
+
+func TestBrowserPushNotificationConstant(t *testing.T) {
+	// Test that the constant is defined and has the expected value
+	if BrowserPushNotification == "" {
+		t.Error("BrowserPushNotification constant should not be empty")
+	}
+
+	expectedValue := "BROWSER_PUSH_NOTIFICATION"
+	if BrowserPushNotification != expectedValue {
+		t.Errorf("Expected BrowserPushNotification to be %s, got %s", expectedValue, BrowserPushNotification)
+	}
+}
+
+func TestNotificationRepositoryInterface(t *testing.T) {
+	// Test that our implementation satisfies the interface
+	var _ NotificationsRepository = &SqlNotificationsRepository{}
+
+	// Test that we can get the repository instance
+	repo := Get()
+	if repo == nil {
+		t.Error("Get() should return a non-nil repository instance")
+	}
+
+	// Test that New creates a valid repository
+	newRepo := New(testDatastore)
+	if newRepo == nil {
+		t.Error("New() should return a non-nil repository instance")
+	}
+}
+
+func TestAddDuplicateNotification(t *testing.T) {
+	channel := "DUPLICATE_TEST_CHANNEL"
+	destination := "duplicate-dest"
+
+	// Add the same notification twice
+	err := testRepo.AddNotification(channel, destination)
+	if err != nil {
+		t.Errorf("Failed to add notification first time: %v", err)
+	}
+
+	err = testRepo.AddNotification(channel, destination)
+	if err != nil {
+		t.Errorf("Failed to add duplicate notification: %v", err)
+	}
+
+	// Check how many destinations we have (should handle duplicates gracefully)
+	destinations, err := testRepo.GetNotificationDestinationsForChannel(channel)
+	if err != nil {
+		t.Errorf("Failed to get notification destinations: %v", err)
+	}
+
+	// The behavior for duplicates depends on database constraints
+	// This test documents the current behavior
+	if len(destinations) == 0 {
+		t.Error("Should have at least one destination even with duplicates")
+	}
+}
+
+func TestChannelIsolation(t *testing.T) {
+	channel1 := "CHANNEL_1"
+	channel2 := "CHANNEL_2"
+	destination1 := "dest-for-channel-1"
+	destination2 := "dest-for-channel-2"
+
+	// Add notifications to different channels
+	err := testRepo.AddNotification(channel1, destination1)
+	if err != nil {
+		t.Errorf("Failed to add notification to channel1: %v", err)
+	}
+
+	err = testRepo.AddNotification(channel2, destination2)
+	if err != nil {
+		t.Errorf("Failed to add notification to channel2: %v", err)
+	}
+
+	// Verify channel isolation
+	destinations1, err := testRepo.GetNotificationDestinationsForChannel(channel1)
+	if err != nil {
+		t.Errorf("Failed to get destinations for channel1: %v", err)
+	}
+
+	destinations2, err := testRepo.GetNotificationDestinationsForChannel(channel2)
+	if err != nil {
+		t.Errorf("Failed to get destinations for channel2: %v", err)
+	}
+
+	// Each channel should only have its own destination
+	if len(destinations1) != 1 || destinations1[0] != destination1 {
+		t.Errorf("Channel1 should only have destination1, got: %v", destinations1)
+	}
+
+	if len(destinations2) != 1 || destinations2[0] != destination2 {
+		t.Errorf("Channel2 should only have destination2, got: %v", destinations2)
+	}
+
+	// Remove from one channel shouldn't affect the other
+	err = testRepo.RemoveNotificationForChannel(channel1, destination1)
+	if err != nil {
+		t.Errorf("Failed to remove notification from channel1: %v", err)
+	}
+
+	// Verify channel2 is unaffected
+	destinations2After, err := testRepo.GetNotificationDestinationsForChannel(channel2)
+	if err != nil {
+		t.Errorf("Failed to get destinations for channel2 after removal: %v", err)
+	}
+
+	if len(destinations2After) != 1 || destinations2After[0] != destination2 {
+		t.Errorf("Channel2 should still have destination2 after channel1 removal, got: %v", destinations2After)
+	}
+}
+
+// Benchmark tests
+func BenchmarkAddNotification(b *testing.B) {
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		channel := fmt.Sprintf("BENCH_CHANNEL_%d", i)
+		destination := fmt.Sprintf("bench_destination_%d", i)
+		err := testRepo.AddNotification(channel, destination)
+		if err != nil {
+			b.Fatalf("Failed to add notification: %v", err)
+		}
+	}
+}
+
+func BenchmarkGetNotificationDestinationsForChannel(b *testing.B) {
+	// Setup test data
+	channel := "BENCH_GET_CHANNEL"
+	for i := 0; i < 100; i++ {
+		destination := fmt.Sprintf("bench_get_destination_%d", i)
+		testRepo.AddNotification(channel, destination)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := testRepo.GetNotificationDestinationsForChannel(channel)
+		if err != nil {
+			b.Fatalf("Failed to get notification destinations: %v", err)
+		}
+	}
+}
+
+func BenchmarkRemoveNotificationForChannel(b *testing.B) {
+	// Setup test data
+	channel := "BENCH_REMOVE_CHANNEL"
+	destinations := make([]string, b.N)
+	for i := 0; i < b.N; i++ {
+		destination := fmt.Sprintf("bench_remove_destination_%d", i)
+		destinations[i] = destination
+		testRepo.AddNotification(channel, destination)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		err := testRepo.RemoveNotificationForChannel(channel, destinations[i])
+		if err != nil {
+			b.Fatalf("Failed to remove notification: %v", err)
+		}
+	}
+}

services/notifications/service.go

@@ -0,0 +1,129 @@
+package notifications
+
+import (
+	"fmt"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/notifications/browser"
+	"github.com/owncast/owncast/notifications/discord"
+	"github.com/owncast/owncast/persistence/configrepository"
+	"github.com/owncast/owncast/persistence/notificationsrepository"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+// Service defines the interface for notification operations.
+type Service interface {
+	Notify()
+}
+
+// notificationService handles notification dispatching and channel management.
+type notificationService struct {
+	repository       notificationsrepository.NotificationsRepository
+	configRepository configrepository.ConfigRepository
+	browser          *browser.Browser
+	discord          *discord.Discord
+}
+
+// New creates a new instance of the notification service.
+func New(datastore *data.Datastore) (Service, error) {
+	service := &notificationService{
+		repository:       notificationsrepository.New(datastore),
+		configRepository: configrepository.Get(),
+	}
+
+	if err := service.setupBrowserPush(datastore); err != nil {
+		log.Error(err)
+	}
+	if err := service.setupDiscord(); err != nil {
+		log.Error(err)
+	}
+
+	return service, nil
+}
+
+func (s *notificationService) setupBrowserPush(datastore *data.Datastore) error {
+	if s.configRepository.GetBrowserPushConfig().Enabled {
+		publicKey, err := s.configRepository.GetBrowserPushPublicKey()
+		if err != nil || publicKey == "" {
+			return errors.Wrap(err, "browser notifier disabled, failed to get browser push public key")
+		}
+
+		privateKey, err := s.configRepository.GetBrowserPushPrivateKey()
+		if err != nil || privateKey == "" {
+			return errors.Wrap(err, "browser notifier disabled, failed to get browser push private key")
+		}
+
+		browserNotifier, err := browser.New(datastore, publicKey, privateKey)
+		if err != nil {
+			return errors.Wrap(err, "error creating browser notifier")
+		}
+		s.browser = browserNotifier
+	}
+	return nil
+}
+
+func (s *notificationService) setupDiscord() error {
+	discordConfig := s.configRepository.GetDiscordConfig()
+	if discordConfig.Enabled && discordConfig.Webhook != "" {
+		var image string
+		if serverURL := s.configRepository.GetServerURL(); serverURL != "" {
+			image = serverURL + "/logo"
+		}
+		discordNotifier, err := discord.New(
+			s.configRepository.GetServerName(),
+			image,
+			discordConfig.Webhook,
+		)
+		if err != nil {
+			return errors.Wrap(err, "error creating discord notifier")
+		}
+		s.discord = discordNotifier
+	}
+	return nil
+}
+
+func (s *notificationService) notifyBrowserPush() {
+	destinations, err := s.repository.GetNotificationDestinationsForChannel(notificationsrepository.BrowserPushNotification)
+	if err != nil {
+		log.Errorln("error getting browser push notification destinations", err)
+	}
+	for _, destination := range destinations {
+		unsubscribed, err := s.browser.Send(destination, s.configRepository.GetServerName(), s.configRepository.GetBrowserPushConfig().GoLiveMessage)
+		if unsubscribed {
+			// If the error is "unsubscribed", then remove the destination from the database.
+			if err := s.repository.RemoveNotificationForChannel(notificationsrepository.BrowserPushNotification, destination); err != nil {
+				log.Errorln(err)
+			}
+		} else if err != nil {
+			log.Errorln(err)
+		}
+	}
+}
+
+func (s *notificationService) notifyDiscord() {
+	goLiveMessage := s.configRepository.GetDiscordConfig().GoLiveMessage
+	streamTitle := s.configRepository.GetStreamTitle()
+	if streamTitle != "" {
+		goLiveMessage += "\n" + streamTitle
+	}
+	message := fmt.Sprintf("%s\n\n%s", goLiveMessage, s.configRepository.GetServerURL())
+
+	if err := s.discord.Send(message); err != nil {
+		log.Errorln("error sending discord message", err)
+	}
+}
+
+// Notify will fire the different notification channels.
+func (s *notificationService) Notify() {
+	if s.browser != nil {
+		s.notifyBrowserPush()
+	}
+
+	if s.discord != nil {
+		s.notifyDiscord()
+	}
+}
+
+// Compile-time verification that notificationService implements Service.
+var _ Service = (*notificationService)(nil)

webserver/handlers/notifications.go

@@ -5,10 +5,8 @@ import (
 	"net/http"
 
 	"github.com/owncast/owncast/models"
-	"github.com/owncast/owncast/notifications"
+	notificationsrepo "github.com/owncast/owncast/persistence/notificationsrepository"
-
 	"github.com/owncast/owncast/utils"
-
 	webutils "github.com/owncast/owncast/webserver/utils"
 	log "github.com/sirupsen/logrus"
 )
@@ -37,14 +35,14 @@ func RegisterForLiveNotifications(u models.User, w http.ResponseWriter, r *http.
 	}
 
 	// Make sure the requested channel is one we want to handle.
-	validTypes := []string{notifications.BrowserPushNotification}
+	validTypes := []string{notificationsrepo.BrowserPushNotification}
 	_, validChannel := utils.FindInSlice(validTypes, req.Channel)
 	if !validChannel {
 		webutils.WriteSimpleResponse(w, false, "invalid notification channel: "+req.Channel)
 		return
 	}
 
-	if err := notifications.AddNotification(req.Channel, req.Destination); err != nil {
+	if err := notificationsrepo.Get().AddNotification(req.Channel, req.Destination); err != nil {
 		log.Errorln(err)
 		webutils.WriteSimpleResponse(w, false, "unable to save notification")
 		return