diff options
| author | Gabe Kangas <gabek@real-ity.com> | 2023-07-21 22:25:59 -0700 |
|---|---|---|
| committer | Gabe Kangas <gabek@real-ity.com> | 2023-08-12 20:35:36 -0700 |
| commit | c572fb76cedc81e65f0b85c3b7aff373c4959146 (patch) | |
| tree | c69512112b8652c7aad9bf0fbb546c74b8347ba9 | |
| parent | 659a19bf2c98720d0fba25104ffb51190ca19278 (diff) | |
161 files changed, 2300 insertions, 1803 deletions
diff --git a/activitypub/activitypub.go b/activitypub/activitypub.go deleted file mode 100644 index 4b3dedaca..000000000 --- a/activitypub/activitypub.go +++ /dev/null @@ -1,61 +0,0 @@ -package activitypub - -import ( - "net/http" - - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/inbox" - "github.com/owncast/owncast/activitypub/outbox" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/workerpool" - "github.com/owncast/owncast/storage/configrepository" - "github.com/owncast/owncast/storage/data" - - "github.com/owncast/owncast/models" - log "github.com/sirupsen/logrus" -) - -var configRepository = configrepository.Get() - -// Start will initialize and start the federation support. -func Start(datastore *data.Store, router *http.ServeMux) { - persistence.Setup(datastore) - workerpool.InitOutboundWorkerPool() - inbox.InitInboxWorkerPool() - StartRouter(router) - - // Generate the keys for signing federated activity if needed. - if configRepository.GetPrivateKey() == "" { - privateKey, publicKey, err := crypto.GenerateKeys() - _ = configRepository.SetPrivateKey(string(privateKey)) - _ = configRepository.SetPublicKey(string(publicKey)) - if err != nil { - log.Errorln("Unable to get private key", err) - } - } -} - -// SendLive will send a "Go Live" message to followers. -func SendLive() error { - return outbox.SendLive() -} - -// SendPublicFederatedMessage will send an arbitrary provided message to followers. -func SendPublicFederatedMessage(message string) error { - return outbox.SendPublicMessage(message) -} - -// SendDirectFederatedMessage will send a direct message to a single account. -func SendDirectFederatedMessage(message, account string) error { - return outbox.SendDirectMessageToAccount(message, account) -} - -// GetFollowerCount will return the local tracked follower count. -func GetFollowerCount() (int64, error) { - return persistence.GetFollowerCount() -} - -// GetPendingFollowRequests will return the pending follow requests. -func GetPendingFollowRequests() ([]models.Follower, error) { - return persistence.GetPendingFollowRequests() -} diff --git a/activitypub/inbox/chat.go b/activitypub/inbox/chat.go deleted file mode 100644 index 70b1c3fef..000000000 --- a/activitypub/inbox/chat.go +++ /dev/null @@ -1,64 +0,0 @@ -package inbox - -import ( - "fmt" - - "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/resolvers" - "github.com/owncast/owncast/core/chat" - "github.com/owncast/owncast/core/chat/events" - "github.com/owncast/owncast/storage/configrepository" -) - -var configRepository = configrepository.Get() - -func handleEngagementActivity(eventType events.EventType, isLiveNotification bool, actorReference vocab.ActivityStreamsActorProperty, action string) error { - // Do nothing if displaying engagement actions has been turned off. - if !configRepository.GetFederationShowEngagement() { - return nil - } - - // Do nothing if chat is disabled - if configRepository.GetChatDisabled() { - return nil - } - - // Get actor of the action - actor, _ := resolvers.GetResolvedActorFromActorProperty(actorReference) - - // Send chat message - actorName := actor.Name - if actorName == "" { - actorName = actor.Username - } - actorIRI := actorReference.Begin().GetIRI().String() - - userPrefix := fmt.Sprintf("%s ", actorName) - var suffix string - if isLiveNotification && action == events.FediverseEngagementLike { - suffix = "liked that this stream went live." - } else if action == events.FediverseEngagementLike { - suffix = fmt.Sprintf("liked a post from %s.", configRepository.GetServerName()) - } else if isLiveNotification && action == events.FediverseEngagementRepost { - suffix = "shared this stream with their followers." - } else if action == events.FediverseEngagementRepost { - suffix = fmt.Sprintf("shared a post from %s.", configRepository.GetServerName()) - } else if action == events.FediverseEngagementFollow { - suffix = "followed this stream." - } else { - return fmt.Errorf("could not handle event for sending to chat: %s", action) - } - body := fmt.Sprintf("%s %s", userPrefix, suffix) - - var image *string - if actor.Image != nil { - s := actor.Image.String() - image = &s - } - - if err := chat.SendFediverseAction(eventType, actor.FullUsername, image, body, actorIRI); err != nil { - return err - } - - return nil -} diff --git a/activitypub/inbox/update.go b/activitypub/inbox/update.go deleted file mode 100644 index 047ed3398..000000000 --- a/activitypub/inbox/update.go +++ /dev/null @@ -1,25 +0,0 @@ -package inbox - -import ( - "context" - - "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/resolvers" - log "github.com/sirupsen/logrus" -) - -func handleUpdateRequest(c context.Context, activity vocab.ActivityStreamsUpdate) error { - // We only care about update events to followers. - if !activity.GetActivityStreamsObject().At(0).IsActivityStreamsPerson() { - return nil - } - - actor, err := resolvers.GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) - if err != nil { - log.Errorln(err) - return err - } - - return persistence.UpdateFollower(actor.ActorIri.String(), actor.Inbox.String(), actor.Name, actor.FullUsername, actor.Image.String()) -} diff --git a/activitypub/router.go b/activitypub/router.go deleted file mode 100644 index 0d13b1115..000000000 --- a/activitypub/router.go +++ /dev/null @@ -1,35 +0,0 @@ -package activitypub - -import ( - "net/http" - - "github.com/owncast/owncast/activitypub/controllers" - "github.com/owncast/owncast/webserver/middleware" -) - -// StartRouter will start the federation specific http router. -func StartRouter(router *http.ServeMux) { - // WebFinger - router.HandleFunc("/.well-known/webfinger", controllers.WebfingerHandler) - - // Host Metadata - router.HandleFunc("/.well-known/host-meta", controllers.HostMetaController) - - // Nodeinfo v1 - router.HandleFunc("/.well-known/nodeinfo", controllers.NodeInfoController) - - // x-nodeinfo v2 - router.HandleFunc("/.well-known/x-nodeinfo2", controllers.XNodeInfo2Controller) - - // Nodeinfo v2 - router.HandleFunc("/nodeinfo/2.0", controllers.NodeInfoV2Controller) - - // Instance details - router.HandleFunc("/api/v1/instance", controllers.InstanceV1Controller) - - // Single ActivityPub Actor - router.HandleFunc("/federation/user/", middleware.RequireActivityPubOrRedirect(controllers.ActorHandler)) - - // Single AP object - router.HandleFunc("/federation/", middleware.RequireActivityPubOrRedirect(controllers.ObjectHandler)) -} diff --git a/cmd/application.go b/cmd/application.go new file mode 100644 index 000000000..93afaed81 --- /dev/null +++ b/cmd/application.go @@ -0,0 +1,48 @@ +package cmd + +import ( + "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/services/metrics" + "github.com/owncast/owncast/storage/configrepository" + log "github.com/sirupsen/logrus" +) + +type Application struct { + configservice *config.Config + metricsservice *metrics.Metrics + configRepository *configrepository.SqlConfigRepository + + maximumConcurrentConnectionLimit int64 +} + +/* +The order of this setup matters. +- Parse flags +- Set the session runtime values +- Use the session values to configure data persistence +*/ +func (app *Application) Start() { + app.configservice = config.Get() + + app.parseFlags() + app.configureLogging(*enableDebugOptions, *enableVerboseLogging, app.configservice.LogDirectory) + app.showStartupMessage() + + app.setSessionConfig() + app.createDirectories() + + app.maximumConcurrentConnectionLimit = getMaximumConcurrentConnectionLimit() + setSystemConcurrentConnectionLimit(app.maximumConcurrentConnectionLimit) + + // If we're restoring a backup, do that and exit. + if *restoreDatabaseFile != "" { + app.handleRestoreBackup(restoreDatabaseFile) + log.Exit(0) + } + + if *backupDirectory != "" { + app.configservice.BackupDirectory = *backupDirectory + } + + app.startServices() +} diff --git a/cmd/backuprestore.go b/cmd/backuprestore.go new file mode 100644 index 000000000..1d10cf2aa --- /dev/null +++ b/cmd/backuprestore.go @@ -0,0 +1,20 @@ +package cmd + +import ( + "github.com/owncast/owncast/utils" + log "github.com/sirupsen/logrus" +) + +func (app *Application) handleRestoreBackup(restoreDatabaseFile *string) { + // Allows a user to restore a specific database backup + databaseFile := app.configservice.DatabaseFilePath + if *dbFile != "" { + databaseFile = *dbFile + } + + if err := utils.Restore(*restoreDatabaseFile, databaseFile); err != nil { + log.Fatalln(err) + } + + log.Println("Database has been restored. Restart Owncast.") +} diff --git a/core/chat/concurrentConnections.go b/cmd/concurrentConnections.go index cbbea81c3..03b053cb9 100644 --- a/core/chat/concurrentConnections.go +++ b/cmd/concurrentConnections.go @@ -2,7 +2,7 @@ //go:build !freebsd && !windows // +build !freebsd,!windows -package chat +package cmd import ( "syscall" @@ -24,3 +24,15 @@ func setSystemConcurrentConnectionLimit(limit int64) { log.Traceln("Max process connection count changed from system limit of", originalLimit, "to", limit) } + +func getMaximumConcurrentConnectionLimit() int64 { + var rLimit syscall.Rlimit + if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil { + log.Fatalln(err) + } + + // Return the limit to 70% of max so the machine doesn't die even if it's maxed out for some reason. + proposedLimit := int64(float32(rLimit.Max) * 0.7) + + return proposedLimit +} diff --git a/core/chat/concurrentConnections_freebsd.go b/cmd/concurrentConnections_freebsd.go index 68d4955b3..d948c01eb 100644 --- a/core/chat/concurrentConnections_freebsd.go +++ b/cmd/concurrentConnections_freebsd.go @@ -1,7 +1,7 @@ //go:build freebsd // +build freebsd -package chat +package cmd import ( "syscall" diff --git a/core/chat/concurrentConnections_windows.go b/cmd/concurrentConnections_windows.go index ef61843b3..7ccbb51de 100644 --- a/core/chat/concurrentConnections_windows.go +++ b/cmd/concurrentConnections_windows.go @@ -1,6 +1,6 @@ //go:build windows // +build windows -package chat +package cmd func setSystemConcurrentConnectionLimit(limit int64) {} diff --git a/cmd/config.go b/cmd/config.go new file mode 100644 index 000000000..178ef7868 --- /dev/null +++ b/cmd/config.go @@ -0,0 +1,71 @@ +package cmd + +import ( + "strconv" + + "github.com/owncast/owncast/storage/configrepository" + log "github.com/sirupsen/logrus" +) + +func (app *Application) setSessionConfig() { + // Stream key + if *newStreamKey != "" { + log.Println("Temporary stream key is set for this session.") + app.configservice.TemporaryStreamKey = *newStreamKey + } + + app.configservice.EnableDebugFeatures = *enableDebugOptions + + if *dbFile != "" { + app.configservice.DatabaseFilePath = *dbFile + } + + if *logDirectory != "" { + app.configservice.LogDirectory = *logDirectory + } +} + +func (app *Application) saveUpdatedConfig() { + configRepository := configrepository.Get() + + if *newAdminPassword != "" { + if err := configRepository.SetAdminPassword(*newAdminPassword); err != nil { + log.Errorln("Error setting your admin password.", err) + log.Exit(1) + } else { + log.Infoln("Admin password changed") + } + } + + // Set the web server port + if *webServerPortOverride != "" { + portNumber, err := strconv.Atoi(*webServerPortOverride) + if err != nil { + log.Warnln(err) + return + } + + log.Println("Saving new web server port number to", portNumber) + if err := configRepository.SetHTTPPortNumber(float64(portNumber)); err != nil { + log.Errorln(err) + } + } + app.configservice.WebServerPort = configRepository.GetHTTPPortNumber() + + // Set the web server ip + if *webServerIPOverride != "" { + log.Println("Saving new web server listen IP address to", *webServerIPOverride) + if err := configRepository.SetHTTPListenAddress(*webServerIPOverride); err != nil { + log.Errorln(err) + } + } + app.configservice.WebServerIP = configRepository.GetHTTPListenAddress() + + // Set the rtmp server port + if *rtmpPortOverride > 0 { + log.Println("Saving new RTMP server port number to", *rtmpPortOverride) + if err := configRepository.SetRTMPPortNumber(float64(*rtmpPortOverride)); err != nil { + log.Errorln(err) + } + } +} diff --git a/cmd/console.go b/cmd/console.go new file mode 100644 index 000000000..1a3ddb059 --- /dev/null +++ b/cmd/console.go @@ -0,0 +1,7 @@ +package cmd + +import log "github.com/sirupsen/logrus" + +func (app *Application) showStartupMessage() { + log.Infoln(app.configservice.GetReleaseString()) +} diff --git a/cmd/data.go b/cmd/data.go new file mode 100644 index 000000000..a21ed05c4 --- /dev/null +++ b/cmd/data.go @@ -0,0 +1,4 @@ +package cmd + +func initializeData() { +} diff --git a/cmd/flags.go b/cmd/flags.go new file mode 100644 index 000000000..c99c3bea4 --- /dev/null +++ b/cmd/flags.go @@ -0,0 +1,23 @@ +package cmd + +import ( + "flag" +) + +var ( + dbFile = flag.String("database", "", "Path to the database file.") + logDirectory = flag.String("logdir", "", "Directory where logs will be written to") + backupDirectory = flag.String("backupdir", "", "Directory where backups will be written to") + enableDebugOptions = flag.Bool("enableDebugFeatures", false, "Enable additional debugging options.") + enableVerboseLogging = flag.Bool("enableVerboseLogging", false, "Enable additional logging.") + restoreDatabaseFile = flag.String("restoreDatabase", "", "Restore an Owncast database backup") + newAdminPassword = flag.String("adminpassword", "", "Set your admin password") + newStreamKey = flag.String("streamkey", "", "Set a temporary stream key for this session") + webServerPortOverride = flag.String("webserverport", "", "Force the web server to listen on a specific port") + webServerIPOverride = flag.String("webserverip", "", "Force web server to listen on this IP address") + rtmpPortOverride = flag.Int("rtmpport", 0, "Set listen port for the RTMP server") +) + +func (app *Application) parseFlags() { + flag.Parse() +} diff --git a/cmd/services.go b/cmd/services.go new file mode 100644 index 000000000..24144ba3c --- /dev/null +++ b/cmd/services.go @@ -0,0 +1,4 @@ +package cmd + +func (app *Application) startServices() { +} diff --git a/cmd/setup.go b/cmd/setup.go new file mode 100644 index 000000000..2ed8b314d --- /dev/null +++ b/cmd/setup.go @@ -0,0 +1,162 @@ +package cmd + +import ( + "fmt" + "io" + "io/fs" + "os" + "path" + "path/filepath" + + "github.com/owncast/owncast/logging" + "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/static" + "github.com/owncast/owncast/utils" + "github.com/pkg/errors" + log "github.com/sirupsen/logrus" +) + +func (app *Application) createDirectories() { + // Create the data directory if needed + if !utils.DoesFileExists("data") { + if err := os.Mkdir("./data", 0o700); err != nil { + log.Fatalln("Cannot create data directory", err) + } + } + + // Recreate the temp dir + if utils.DoesFileExists(app.configservice.TempDir) { + err := os.RemoveAll(app.configservice.TempDir) + if err != nil { + log.Fatalln("Unable to remove temp dir! Check permissions.", app.configservice.TempDir, err) + } + } + if err := os.Mkdir(app.configservice.TempDir, 0o700); err != nil { + log.Fatalln("Unable to create temp dir!", err) + } +} + +func (app *Application) configureLogging(enableDebugFeatures bool, enableVerboseLogging bool, logDirectory string) { + logging.Setup(enableDebugFeatures, enableVerboseLogging, logDirectory) + log.SetFormatter(&log.TextFormatter{ + FullTimestamp: true, + }) +} + +// setupEmojiDirectory sets up the custom emoji directory by copying all built-in +// emojis if the directory does not yet exist. +func (app *Application) setupEmojiDirectory() (err error) { + type emojiDirectory struct { + path string + isDir bool + } + + // Migrate old (pre 0.1.0) emoji to new location if they exist. + app.migrateCustomEmojiLocations() + + if utils.DoesFileExists(app.configservice.CustomEmojiPath) { + return nil + } + + if err = os.MkdirAll(app.configservice.CustomEmojiPath, 0o750); err != nil { + return fmt.Errorf("unable to create custom emoji directory: %w", err) + } + + staticFS := static.GetEmoji() + files := []emojiDirectory{} + + walkFunction := func(path string, d os.DirEntry, err error) error { + if path == "." { + return nil + } + + if d.Name() == "LICENSE.md" { + return nil + } + + files = append(files, emojiDirectory{path: path, isDir: d.IsDir()}) + return nil + } + + if err := fs.WalkDir(staticFS, ".", walkFunction); err != nil { + log.Errorln("unable to fetch emojis: " + err.Error()) + return errors.Wrap(err, "unable to fetch embedded emoji files") + } + + if err != nil { + return fmt.Errorf("unable to read built-in emoji files: %w", err) + } + + // Now copy all built-in emojis to the custom emoji directory + for _, path := range files { + emojiPath := filepath.Join(app.configservice.CustomEmojiPath, path.path) + + if path.isDir { + if err := os.Mkdir(emojiPath, 0o700); err != nil { + return errors.Wrap(err, "unable to create emoji directory, check permissions?: "+path.path) + } + continue + } + + memFile, staticOpenErr := staticFS.Open(path.path) + if staticOpenErr != nil { + return errors.Wrap(staticOpenErr, "unable to open emoji file from embedded filesystem") + } + + // nolint:gosec + diskFile, err := os.Create(emojiPath) + if err != nil { + return fmt.Errorf("unable to create custom emoji file on disk: %w", err) + } + + if err != nil { + _ = diskFile.Close() + return fmt.Errorf("unable to open built-in emoji file: %w", err) + } + + if _, err = io.Copy(diskFile, memFile); err != nil { + _ = diskFile.Close() + _ = os.Remove(emojiPath) + return fmt.Errorf("unable to copy built-in emoji file to disk: %w", err) + } + + if err = diskFile.Close(); err != nil { + _ = os.Remove(emojiPath) + return fmt.Errorf("unable to close custom emoji file on disk: %w", err) + } + } + + return nil +} + +// MigrateCustomEmojiLocations migrates custom emoji from the old location to the new location. +func (app *Application) migrateCustomEmojiLocations() { + oldLocation := path.Join("webroot", "img", "emoji") + newLocation := path.Join("data", "emoji") + + if !utils.DoesFileExists(oldLocation) { + return + } + + log.Println("Moving custom emoji directory from", oldLocation, "to", newLocation) + + if err := utils.Move(oldLocation, newLocation); err != nil { + log.Errorln("error moving custom emoji directory", err) + } +} + +func (app *Application) resetDirectories() { + log.Trace("Resetting file directories to a clean slate.") + + // Wipe hls data directory + utils.CleanupDirectory(app.configservice.HLSStoragePath) + + // Remove the previous thumbnail + logo := app.configRepository.GetLogoPath() + if utils.DoesFileExists(logo) { + err := utils.Copy(path.Join("data", logo), filepath.Join(config.DataDirectory, "thumbnail.jpg")) + if err != nil { + log.Warnln(err) + } + } +} diff --git a/core/chat/utils_windows.go b/cmd/utils_windows.go index 56fd59854..56fd59854 100644 --- a/core/chat/utils_windows.go +++ b/cmd/utils_windows.go diff --git a/core/chat/chat.go b/core/chat/chat.go deleted file mode 100644 index 6b27d4a85..000000000 --- a/core/chat/chat.go +++ /dev/null @@ -1,187 +0,0 @@ -package chat - -import ( - "errors" - "net/http" - "sort" - - "github.com/owncast/owncast/core/chat/events" - "github.com/owncast/owncast/models" - "github.com/owncast/owncast/services/config" - "github.com/owncast/owncast/storage/configrepository" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" - log "github.com/sirupsen/logrus" -) - -var ( - getStatus func() models.Status - chatMessagesSentCounter prometheus.Gauge -) - -var configRepository = configrepository.Get() - -// Start begins the chat server. -func Start(getStatusFunc func() models.Status) error { - setupPersistence() - - getStatus = getStatusFunc - _server = NewChat() - - go _server.Run() - - log.Traceln("Chat server started with max connection count of", _server.maxSocketConnectionLimit) - c := config.GetConfig() - - chatMessagesSentCounter = promauto.NewGauge(prometheus.GaugeOpts{ - Name: "total_chat_message_count", - Help: "The number of chat messages incremented over time.", - ConstLabels: map[string]string{ - "version": c.VersionNumber, - "host": configRepository.GetServerURL(), - }, - }) - - return nil -} - -// GetClientsForUser will return chat connections that are owned by a specific user. -func GetClientsForUser(userID string) ([]*Client, error) { - _server.mu.Lock() - defer _server.mu.Unlock() - - clients := map[string][]*Client{} - - for _, client := range _server.clients { - clients[client.User.ID] = append(clients[client.User.ID], client) - } - - if _, exists := clients[userID]; !exists { - return nil, errors.New("no connections for user found") - } - - return clients[userID], nil -} - -// FindClientByID will return a single connected client by ID. -func FindClientByID(clientID uint) (*Client, bool) { - client, found := _server.clients[clientID] - return client, found -} - -// GetClients will return all the current chat clients connected. -func GetClients() []*Client { - clients := []*Client{} - - if _server == nil { - return clients - } - - // Convert the keyed map to a slice. - for _, client := range _server.clients { - clients = append(clients, client) - } - - sort.Slice(clients, func(i, j int) bool { - return clients[i].ConnectedAt.Before(clients[j].ConnectedAt) - }) - - return clients -} - -// SendSystemMessage will send a message string as a system message to all clients. -func SendSystemMessage(text string, ephemeral bool) error { - message := events.SystemMessageEvent{ - MessageEvent: events.MessageEvent{ - Body: text, - }, - } - message.SetDefaults() - message.RenderBody() - - if err := Broadcast(&message); err != nil { - log.Errorln("error sending system message", err) - } - - if !ephemeral { - saveEvent(message.ID, nil, message.Body, message.GetMessageType(), nil, message.Timestamp, nil, nil, nil, nil) - } - - return nil -} - -// SendFediverseAction will send a message indicating some Fediverse engagement took place. -func SendFediverseAction(eventType string, userAccountName string, image *string, body string, link string) error { - message := events.FediverseEngagementEvent{ - Event: events.Event{ - Type: eventType, - }, - MessageEvent: events.MessageEvent{ - Body: body, - }, - UserAccountName: userAccountName, - Image: image, - Link: link, - } - - message.SetDefaults() - message.RenderBody() - - if err := Broadcast(&message); err != nil { - log.Errorln("error sending system message", err) - return err - } - - saveFederatedAction(message) - - return nil -} - -// SendSystemAction will send a system action string as an action event to all clients. -func SendSystemAction(text string, ephemeral bool) error { - message := events.ActionEvent{ - MessageEvent: events.MessageEvent{ - Body: text, - }, - } - - message.SetDefaults() - message.RenderBody() - - if err := Broadcast(&message); err != nil { - log.Errorln("error sending system chat action") - } - - if !ephemeral { - saveEvent(message.ID, nil, message.Body, message.GetMessageType(), nil, message.Timestamp, nil, nil, nil, nil) - } - - return nil -} - -// SendAllWelcomeMessage will send the chat message to all connected clients. -func SendAllWelcomeMessage() { - _server.sendAllWelcomeMessage() -} - -// SendSystemMessageToClient will send a single message to a single connected chat client. -func SendSystemMessageToClient(clientID uint, text string) { - if client, foundClient := FindClientByID(clientID); foundClient { - _server.sendSystemMessageToClient(client, text) - } -} - -// Broadcast will send all connected clients the outbound object provided. -func Broadcast(event events.OutboundEvent) error { - return _server.Broadcast(event.GetBroadcastPayload()) -} - -// HandleClientConnection handles a single inbound websocket connection. -func HandleClientConnection(w http.ResponseWriter, r *http.Request) { - _server.HandleClientConnection(w, r) -} - -// DisconnectClients will forcefully disconnect all clients belonging to a user by ID. -func DisconnectClients(clients []*Client) { - _server.DisconnectClients(clients) -} diff --git a/core/chat/events/userJoinedEvent.go b/core/chat/events/userJoinedEvent.go deleted file mode 100644 index 475b74ce5..000000000 --- a/core/chat/events/userJoinedEvent.go +++ /dev/null @@ -1,17 +0,0 @@ -package events - -// UserJoinedEvent is the event fired when a user joins chat. -type UserJoinedEvent struct { - Event - UserEvent -} - -// GetBroadcastPayload will return the object to send to all chat users. -func (e *UserJoinedEvent) GetBroadcastPayload() EventPayload { - return EventPayload{ - "type": UserJoined, - "id": e.ID, - "timestamp": e.Timestamp, - "user": e.User, - } -} diff --git a/core/chat/utils.go b/core/chat/utils.go deleted file mode 100644 index fae7beb29..000000000 --- a/core/chat/utils.go +++ /dev/null @@ -1,22 +0,0 @@ -//go:build !windows -// +build !windows - -package chat - -import ( - "syscall" - - log "github.com/sirupsen/logrus" -) - -func getMaximumConcurrentConnectionLimit() int64 { - var rLimit syscall.Rlimit - if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil { - log.Fatalln(err) - } - - // Return the limit to 70% of max so the machine doesn't die even if it's maxed out for some reason. - proposedLimit := int64(float32(rLimit.Max) * 0.7) - - return proposedLimit -} diff --git a/core/core.go b/core/core.go index 6c53c33dc..39465030e 100644 --- a/core/core.go +++ b/core/core.go @@ -9,9 +9,9 @@ import ( "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" - "github.com/owncast/owncast/services/auth" "github.com/owncast/owncast/services/config" "github.com/owncast/owncast/services/notifications" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/services/webhooks" "github.com/owncast/owncast/services/yp" "github.com/owncast/owncast/storage/configrepository" @@ -31,11 +31,10 @@ var ( fileWriter = transcoder.FileWriterReceiverService{} ) -var configRepository = configrepository.Get() - // Start starts up the core processing. func Start() error { resetDirectories() + configRepository := configrepository.Get() configRepository.PopulateDefaults() @@ -59,7 +58,7 @@ func Start() error { } // user.SetupUsers() - auth.Setup(data.GetDatastore()) + // auth.Setup(data.GetDatastore()) fileWriter.SetupFileWriterReceiverService(&handler) @@ -68,77 +67,29 @@ func Start() error { return err } - _yp = yp.NewYP(GetStatus) + s := status.Get() + gsf := func() *models.Status { + s := status.Get() + return &s.Status + } + + _yp = yp.NewYP(gsf) - if err := chat.Start(GetStatus); err != nil { + if err := chat.Start(gsf); err != nil { log.Errorln(err) } // start the rtmp server - go rtmp.Start(setStreamAsConnected, setBroadcaster) + go rtmp.Start(setStreamAsConnected, s.SetBroadcaster) rtmpPort := configRepository.GetRTMPPortNumber() if rtmpPort != 1935 { log.Infof("RTMP is accepting inbound streams on port %d.", rtmpPort) } - webhooks.InitTemporarySingleton(GetStatus) + webhooks.InitTemporarySingleton(gsf) notifications.Setup(data.GetDatastore()) return nil } - -func createInitialOfflineState() error { - transitionToOfflineVideoStreamContent() - - return nil -} - -// transitionToOfflineVideoStreamContent will overwrite the current stream with the -// offline video stream state only. No live stream HLS segments will continue to be -// referenced. -func transitionToOfflineVideoStreamContent() { - log.Traceln("Firing transcoder with offline stream state") - - _transcoder := transcoder.NewTranscoder() - _transcoder.SetIdentifier("offline") - _transcoder.SetLatencyLevel(models.GetLatencyLevel(4)) - _transcoder.SetIsEvent(true) - - offlineFilePath, err := saveOfflineClipToDisk("offline.ts") - if err != nil { - log.Fatalln("unable to save offline clip:", err) - } - - _transcoder.SetInput(offlineFilePath) - go _transcoder.Start(false) - - // Copy the logo to be the thumbnail - logo := configRepository.GetLogoPath() - c := config.GetConfig() - dst := filepath.Join(c.TempDir, "thumbnail.jpg") - if err = utils.Copy(filepath.Join("data", logo), dst); err != nil { - log.Warnln(err) - } - - // Delete the preview Gif - _ = os.Remove(path.Join(config.DataDirectory, "preview.gif")) -} - -func resetDirectories() { - log.Trace("Resetting file directories to a clean slate.") - - // Wipe hls data directory - c := config.GetConfig() - utils.CleanupDirectory(c.HLSStoragePath) - - // Remove the previous thumbnail - logo := configRepository.GetLogoPath() - if utils.DoesFileExists(logo) { - err := utils.Copy(path.Join("data", logo), filepath.Join(config.DataDirectory, "thumbnail.jpg")) - if err != nil { - log.Warnln(err) - } - } -} diff --git a/core/offlineState.go b/core/offlineState.go index 19e7d874e..09242032f 100644 --- a/core/offlineState.go +++ b/core/offlineState.go @@ -19,7 +19,7 @@ func appendOfflineToVariantPlaylist(index int, playlistFilePath string) { return } - c := config.GetConfig() + c := config.Get() tmpFileName := fmt.Sprintf("tmp-stream-%d.m3u8", index) atomicWriteTmpPlaylistFile, err := os.CreateTemp(c.TempDir, tmpFileName) if err != nil { @@ -50,7 +50,7 @@ func appendOfflineToVariantPlaylist(index int, playlistFilePath string) { } func makeVariantIndexOffline(index int, offlineFilePath string, offlineFilename string) { - c := config.GetConfig() + c := config.Get() playlistFilePath := fmt.Sprintf(filepath.Join(c.HLSStoragePath, "%d/stream.m3u8"), index) segmentFilePath := fmt.Sprintf(filepath.Join(c.HLSStoragePath, "%d/%s"), index, offlineFilename) @@ -96,7 +96,7 @@ func createEmptyOfflinePlaylist(playlistFilePath string, offlineFilename string) func saveOfflineClipToDisk(offlineFilename string) (string, error) { offlineFileData := static.GetOfflineSegment() - c := config.GetConfig() + c := config.Get() offlineTmpFile, err := os.CreateTemp(c.TempDir, offlineFilename) if err != nil { log.Errorln("unable to create temp file for offline video segment", err) diff --git a/core/stats.go b/core/stats.go index cee571251..a447bbae4 100644 --- a/core/stats.go +++ b/core/stats.go @@ -9,6 +9,7 @@ import ( "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/geoip" + "github.com/owncast/owncast/storage/configrepository" ) var ( @@ -44,6 +45,8 @@ func IsStreamConnected() bool { return false } + configRepository := configrepository.Get() + // Kind of a hack. It takes a handful of seconds between a RTMP connection and when HLS data is available. // So account for that with an artificial buffer of four segments. timeSinceLastConnected := time.Since(_stats.LastConnectTime.Time).Seconds() @@ -110,6 +113,8 @@ func pruneViewerCount() { } func saveStats() { + configRepository := configrepository.Get() + if err := configRepository.SetPeakOverallViewerCount(_stats.OverallMaxViewerCount); err != nil { log.Errorln("error saving viewer count", err) } @@ -124,6 +129,8 @@ func saveStats() { } func getSavedStats() models.Stats { + configRepository := configrepository.Get() + savedLastDisconnectTime, _ := configRepository.GetLastDisconnectTime() result := models.Stats{ diff --git a/core/status.go b/core/status.go deleted file mode 100644 index 6e8542467..000000000 --- a/core/status.go +++ /dev/null @@ -1,46 +0,0 @@ -package core - -import ( - "github.com/owncast/owncast/models" - "github.com/owncast/owncast/services/config" -) - -// GetStatus gets the status of the system. -func GetStatus() models.Status { - if _stats == nil { - return models.Status{} - } - - viewerCount := 0 - if IsStreamConnected() { - viewerCount = len(_stats.Viewers) - } - - c := config.GetConfig() - - return models.Status{ - Online: IsStreamConnected(), - ViewerCount: viewerCount, - OverallMaxViewerCount: _stats.OverallMaxViewerCount, - SessionMaxViewerCount: _stats.SessionMaxViewerCount, - LastDisconnectTime: _stats.LastDisconnectTime, - LastConnectTime: _stats.LastConnectTime, - VersionNumber: c.VersionNumber, - StreamTitle: configRepository.GetStreamTitle(), - } -} - -// GetCurrentBroadcast will return the currently active broadcast. -func GetCurrentBroadcast() *models.CurrentBroadcast { - return _currentBroadcast -} - -// setBroadcaster will store the current inbound broadcasting details. -func setBroadcaster(broadcaster models.Broadcaster) { - _broadcaster = &broadcaster -} - -// GetBroadcaster will return the details of the currently active broadcaster. -func GetBroadcaster() *models.Broadcaster { - return _broadcaster -} diff --git a/core/streamState.go b/core/streamState.go deleted file mode 100644 index 8cd64c97f..000000000 --- a/core/streamState.go +++ /dev/null @@ -1,203 +0,0 @@ -package core - -import ( - "context" - "io" - "time" - - log "github.com/sirupsen/logrus" - - "github.com/owncast/owncast/activitypub" - "github.com/owncast/owncast/core/chat" - "github.com/owncast/owncast/models" - "github.com/owncast/owncast/services/config" - "github.com/owncast/owncast/services/notifications" - "github.com/owncast/owncast/services/webhooks" - "github.com/owncast/owncast/storage/data" - "github.com/owncast/owncast/utils" - "github.com/owncast/owncast/video/rtmp" - "github.com/owncast/owncast/video/transcoder" -) - -// After the stream goes offline this timer fires a full cleanup after N min. -var _offlineCleanupTimer *time.Timer - -// While a stream takes place cleanup old HLS content every N min. -var _onlineCleanupTicker *time.Ticker - -var _currentBroadcast *models.CurrentBroadcast - -var _onlineTimerCancelFunc context.CancelFunc - -var _lastNotified *time.Time - -// setStreamAsConnected sets the stream as connected. -func setStreamAsConnected(rtmpOut *io.PipeReader) { - now := utils.NullTime{Time: time.Now(), Valid: true} - _stats.StreamConnected = true - _stats.LastDisconnectTime = nil - _stats.LastConnectTime = &now - _stats.SessionMaxViewerCount = 0 - - _currentBroadcast = &models.CurrentBroadcast{ - LatencyLevel: configRepository.GetStreamLatencyLevel(), - OutputSettings: configRepository.GetStreamOutputVariants(), - } - - StopOfflineCleanupTimer() - startOnlineCleanupTimer() - - if _yp != nil { - go _yp.Start() - } - - c := config.GetConfig() - segmentPath := c.HLSStoragePath - - if err := setupStorage(); err != nil { - log.Fatalln("failed to setup the storage", err) - } - - go func() { - _transcoder = transcoder.NewTranscoder() - _transcoder.TranscoderCompleted = func(error) { - SetStreamAsDisconnected() - _transcoder = nil - _currentBroadcast = nil - } - _transcoder.SetStdin(rtmpOut) - _transcoder.Start(true) - }() - - webhookManager := webhooks.Get() - go webhookManager.SendStreamStatusEvent(models.StreamStarted) - transcoder.StartThumbnailGenerator(segmentPath, configRepository.FindHighestVideoQualityIndex(_currentBroadcast.OutputSettings)) - - _ = chat.SendSystemAction("Stay tuned, the stream is **starting**!", true) - chat.SendAllWelcomeMessage() - - // Send delayed notification messages. - _onlineTimerCancelFunc = startLiveStreamNotificationsTimer() -} - -// SetStreamAsDisconnected sets the stream as disconnected. -func SetStreamAsDisconnected() { - _ = chat.SendSystemAction("The stream is ending.", true) - - now := utils.NullTime{Time: time.Now(), Valid: true} - if _onlineTimerCancelFunc != nil { - _onlineTimerCancelFunc() - } - - _stats.StreamConnected = false - _stats.LastDisconnectTime = &now - _stats.LastConnectTime = nil - _broadcaster = nil - - offlineFilename := "offline.ts" - - offlineFilePath, err := saveOfflineClipToDisk(offlineFilename) - if err != nil { - log.Errorln(err) - return - } - - transcoder.StopThumbnailGenerator() - rtmp.Disconnect() - - if _yp != nil { - _yp.Stop() - } - - // If there is no current broadcast available the previous stream - // likely failed for some reason. Don't try to append to it. - // Just transition to offline. - if _currentBroadcast == nil { - stopOnlineCleanupTimer() - transitionToOfflineVideoStreamContent() - log.Errorln("unexpected nil _currentBroadcast") - return - } - - for index := range _currentBroadcast.OutputSettings { - makeVariantIndexOffline(index, offlineFilePath, offlineFilename) - } - - StartOfflineCleanupTimer() - stopOnlineCleanupTimer() - saveStats() - - webhookManager := webhooks.Get() - go webhookManager.SendStreamStatusEvent(models.StreamStopped) -} - -// StartOfflineCleanupTimer will fire a cleanup after n minutes being disconnected. -func StartOfflineCleanupTimer() { - _offlineCleanupTimer = time.NewTimer(5 * time.Minute) - go func() { - for range _offlineCleanupTimer.C { - // Set video to offline state - resetDirectories() - transitionToOfflineVideoStreamContent() - } - }() -} - -// StopOfflineCleanupTimer will stop the previous cleanup timer. -func StopOfflineCleanupTimer() { - if _offlineCleanupTimer != nil { - _offlineCleanupTimer.Stop() - } -} - -func startOnlineCleanupTimer() { - _onlineCleanupTicker = time.NewTicker(1 * time.Minute) - go func() { - for range _onlineCleanupTicker.C { - if err := _storage.Cleanup(); err != nil { - log.Errorln(err) - } - } - }() -} - -func stopOnlineCleanupTimer() { - if _onlineCleanupTicker != nil { - _onlineCleanupTicker.Stop() - } -} - -func startLiveStreamNotificationsTimer() context.CancelFunc { - // Send delayed notification messages. - c, cancelFunc := context.WithCancel(context.Background()) - _onlineTimerCancelFunc = cancelFunc - go func(c context.Context) { - select { - case <-time.After(time.Minute * 2.0): - if _lastNotified != nil && time.Since(*_lastNotified) < 10*time.Minute { - return - } - - // Send Fediverse message. - if configRepository.GetFederationEnabled() { - log.Traceln("Sending Federated Go Live message.") - if err := activitypub.SendLive(); err != nil { - log.Errorln(err) - } - } - - // Send notification to those who have registered for them. - if notifier, err := notifications.New(data.GetDatastore()); err != nil { - log.Errorln(err) - } else { - notifier.Notify() - } - - now := time.Now() - _lastNotified = &now - case <-c.Done(): - } - }(c) - - return cancelFunc -} diff --git a/logging/logging.go b/logging/logging.go index 2d5b7d39c..81e01595b 100644 --- a/logging/logging.go +++ b/logging/logging.go @@ -30,20 +30,20 @@ type OCLogger struct { var Logger *OCLogger // Setup configures our custom logging destinations. -func Setup(enableDebugOptions bool, enableVerboseLogging bool) { +func Setup(enableDebugOptions bool, enableVerboseLogging bool, logDirectory string) { // Create the logging directory if needed - loggingDirectory := filepath.Dir(getLogFilePath()) + loggingDirectory := filepath.Dir(logDirectory) if !utils.DoesFileExists(loggingDirectory) { - if err := os.Mkdir(loggingDirectory, 0700); err != nil { + if err := os.Mkdir(loggingDirectory, 0o700); err != nil { logger.Errorln("unable to create logs directory", loggingDirectory, err) } } // Write logs to a file - path := getLogFilePath() + logFile := filepath.Join(logDirectory, "owncast.log") writer, _ := rotatelogs.New( - path+".%Y%m%d%H%M", - rotatelogs.WithLinkName(path), + logFile+".%Y%m%d%H%M", + rotatelogs.WithLinkName(logFile), rotatelogs.WithMaxAge(time.Duration(86400)*time.Second), rotatelogs.WithRotationTime(time.Duration(604800)*time.Second), ) diff --git a/logging/paths.go b/logging/paths.go deleted file mode 100644 index a3b747648..000000000 --- a/logging/paths.go +++ /dev/null @@ -1,18 +0,0 @@ -package logging - -import ( - "path/filepath" - - "github.com/owncast/owncast/services/config" -) - -// GetTranscoderLogFilePath returns the logging path for the transcoder log output. -func GetTranscoderLogFilePath() string { - c := config.GetConfig() - return filepath.Join(c.LogDirectory, "transcoder.log") -} - -func getLogFilePath() string { - c := config.GetConfig() - return filepath.Join(c.LogDirectory, "owncast.log") -} @@ -1,173 +1,43 @@ package main -import ( - "flag" - "os" - "strconv" +import "github.com/owncast/owncast/cmd" - "github.com/owncast/owncast/logging" - "github.com/owncast/owncast/storage/configrepository" - "github.com/owncast/owncast/webserver" - log "github.com/sirupsen/logrus" - - "github.com/owncast/owncast/core" - configservice "github.com/owncast/owncast/services/config" - "github.com/owncast/owncast/services/metrics" - - "github.com/owncast/owncast/utils" -) - -var ( - dbFile = flag.String("database", "", "Path to the database file.") - logDirectory = flag.String("logdir", "", "Directory where logs will be written to") - backupDirectory = flag.String("backupdir", "", "Directory where backups will be written to") - enableDebugOptions = flag.Bool("enableDebugFeatures", false, "Enable additional debugging options.") - enableVerboseLogging = flag.Bool("enableVerboseLogging", false, "Enable additional logging.") - restoreDatabaseFile = flag.String("restoreDatabase", "", "Restore an Owncast database backup") - newAdminPassword = flag.String("adminpassword", "", "Set your admin password") - newStreamKey = flag.String("streamkey", "", "Set a temporary stream key for this session") - webServerPortOverride = flag.String("webserverport", "", "Force the web server to listen on a specific port") - webServerIPOverride = flag.String("webserverip", "", "Force web server to listen on this IP address") - rtmpPortOverride = flag.Int("rtmpport", 0, "Set listen port for the RTMP server") - config *configservice.Config -) - -var configRepository = configrepository.Get() - -// nolint:cyclop func main() { - flag.Parse() - - config = configservice.NewConfig() - - if *logDirectory != "" { - config.LogDirectory = *logDirectory - } - - if *backupDirectory != "" { - config.BackupDirectory = *backupDirectory - } - - // Create the data directory if needed - if !utils.DoesFileExists("data") { - if err := os.Mkdir("./data", 0o700); err != nil { - log.Fatalln("Cannot create data directory", err) - } - } - - // Migrate old (pre 0.1.0) emoji to new location if they exist. - utils.MigrateCustomEmojiLocations() - - // Otherwise save the default emoji to the data directory. - if err := data.SetupEmojiDirectory(); err != nil { - log.Fatalln("Cannot set up emoji directory", err) - } - - // Recreate the temp dir - if utils.DoesFileExists(config.TempDir) { - err := os.RemoveAll(config.TempDir) - if err != nil { - log.Fatalln("Unable to remove temp dir! Check permissions.", config.TempDir, err) - } - } - if err := os.Mkdir(config.TempDir, 0o700); err != nil { - log.Fatalln("Unable to create temp dir!", err) - } - - configureLogging(*enableDebugOptions, *enableVerboseLogging) - log.Infoln(config.GetReleaseString()) - - // Allows a user to restore a specific database backup - if *restoreDatabaseFile != "" { - databaseFile := config.DatabaseFilePath - if *dbFile != "" { - databaseFile = *dbFile - } - - if err := utils.Restore(*restoreDatabaseFile, databaseFile); err != nil { - log.Fatalln(err) - } - - log.Println("Database has been restored. Restart Owncast.") - log.Exit(0) - } - - config.EnableDebugFeatures = *enableDebugOptions - - if *dbFile != "" { - config.DatabaseFilePath = *dbFile - } - - if err := data.SetupPersistence(config.DatabaseFilePath); err != nil { - log.Fatalln("failed to open database", err) - } - - handleCommandLineFlags() + app := &cmd.Application{} + app.Start() +} - // starts the core - if err := core.Start(); err != nil { - log.Fatalln("failed to start the core package", err) - } +// var configRepository = configrepository.Get() - go metrics.Start(core.GetStatus) +// // nolint:cyclop +// func main() { +// flag.Parse() - webserver := webserver.New() - if err := webserver.Start(config.WebServerIP, config.WebServerPort); err != nil { - log.Fatalln("failed to start/run the web server", err) - } -} +// config = configservice.NewConfig() -func handleCommandLineFlags() { - if *newAdminPassword != "" { - if err := configRepository.SetAdminPassword(*newAdminPassword); err != nil { - log.Errorln("Error setting your admin password.", err) - log.Exit(1) - } else { - log.Infoln("Admin password changed") - } - } +// // Otherwise save the default emoji to the data directory. +// if err := data.SetupEmojiDirectory(); err != nil { +// log.Fatalln("Cannot set up emoji directory", err) +// } - if *newStreamKey != "" { - log.Println("Temporary stream key is set for this session.") - config.TemporaryStreamKey = *newStreamKey - } +// if err := data.SetupPersistence(config.DatabaseFilePath); err != nil { +// log.Fatalln("failed to open database", err) +// } - // Set the web server port - if *webServerPortOverride != "" { - portNumber, err := strconv.Atoi(*webServerPortOverride) - if err != nil { - log.Warnln(err) - return - } +// handleCommandLineFlags() - log.Println("Saving new web server port number to", portNumber) - if err := configRepository.SetHTTPPortNumber(float64(portNumber)); err != nil { - log.Errorln(err) - } - } - config.WebServerPort = configRepository.GetHTTPPortNumber() +// // starts the core +// if err := core.Start(); err != nil { +// log.Fatalln("failed to start the core package", err) +// } - // Set the web server ip - if *webServerIPOverride != "" { - log.Println("Saving new web server listen IP address to", *webServerIPOverride) - if err := configRepository.SetHTTPListenAddress(*webServerIPOverride); err != nil { - log.Errorln(err) - } - } - config.WebServerIP = configRepository.GetHTTPListenAddress() +// go metrics.Start(core.GetStatus) - // Set the rtmp server port - if *rtmpPortOverride > 0 { - log.Println("Saving new RTMP server port number to", *rtmpPortOverride) - if err := configRepository.SetRTMPPortNumber(float64(*rtmpPortOverride)); err != nil { - log.Errorln(err) - } - } -} +// webserver := webserver.New() +// if err := webserver.Start(config.WebServerIP, config.WebServerPort); err != nil { +// log.Fatalln("failed to start/run the web server", err) +// } +// } -func configureLogging(enableDebugFeatures bool, enableVerboseLogging bool) { - logging.Setup(enableDebugFeatures, enableVerboseLogging) - log.SetFormatter(&log.TextFormatter{ - FullTimestamp: true, - }) -} +// func handleCommandLineFlags() { +// } diff --git a/core/chat/events/actionEvent.go b/models/actionEvent.go index b251b8511..c8abc1bda 100644 --- a/core/chat/events/actionEvent.go +++ b/models/actionEvent.go @@ -1,4 +1,4 @@ -package events +package models // ActionEvent represents an action that took place, not a chat message. type ActionEvent struct { diff --git a/models/chatAccessScopes.go b/models/chatAccessScopes.go new file mode 100644 index 000000000..2ec754dc2 --- /dev/null +++ b/models/chatAccessScopes.go @@ -0,0 +1,12 @@ +package models + +const ( + // ScopeCanSendChatMessages will allow sending chat messages as itself. + ScopeCanSendChatMessages = "CAN_SEND_MESSAGES" + // ScopeCanSendSystemMessages will allow sending chat messages as the system. + ScopeCanSendSystemMessages = "CAN_SEND_SYSTEM_MESSAGES" + // ScopeHasAdminAccess will allow performing administrative actions on the server. + ScopeHasAdminAccess = "HAS_ADMIN_ACCESS" + + ModeratorScopeKey = "MODERATOR" +) diff --git a/core/chat/events/eventtype.go b/models/chatEventTypes.go index 3bfde1f8a..d7883df37 100644 --- a/core/chat/events/eventtype.go +++ b/models/chatEventTypes.go @@ -1,4 +1,4 @@ -package events +package models // EventType is the type of a websocket event. type EventType = string @@ -14,6 +14,8 @@ const ( UserColorChanged EventType = "COLOR_CHANGE" // VisibiltyUpdate is the event sent when a chat message's visibility changes. VisibiltyUpdate EventType = "VISIBILITY-UPDATE" + // VisibiltyToggled is the event sent when a chat message's visibility changes. + VisibiltyToggled EventType = "VISIBILITY-UPDATE" // PING is a ping message. PING EventType = "PING" // PONG is a pong message. diff --git a/core/chat/events/connectedClientInfo.go b/models/connectedClientInfo.go index 2223af361..55126b3dc 100644 --- a/core/chat/events/connectedClientInfo.go +++ b/models/connectedClientInfo.go @@ -1,9 +1,7 @@ -package events - -import "github.com/owncast/owncast/models" +package models // ConnectedClientInfo represents the information about a connected client. type ConnectedClientInfo struct { Event - User *models.User `json:"user"` + User *User `json:"user"` } diff --git a/models/eventType.go b/models/eventType.go deleted file mode 100644 index dd36295ba..000000000 --- a/models/eventType.go +++ /dev/null @@ -1,29 +0,0 @@ -package models - -// EventType is the type of a websocket event. -type EventType = string - -const ( - // MessageSent is the event sent when a chat event takes place. - MessageSent EventType = "CHAT" - // UserJoined is the event sent when a chat user join action takes place. - UserJoined EventType = "USER_JOINED" - // UserNameChanged is the event sent when a chat username change takes place. - UserNameChanged EventType = "NAME_CHANGE" - // VisibiltyToggled is the event sent when a chat message's visibility changes. - VisibiltyToggled EventType = "VISIBILITY-UPDATE" - // PING is a ping message. - PING EventType = "PING" - // PONG is a pong message. - PONG EventType = "PONG" - // StreamStarted represents a stream started event. - StreamStarted EventType = "STREAM_STARTED" - // StreamStopped represents a stream stopped event. - StreamStopped EventType = "STREAM_STOPPED" - // StreamTitleUpdated is the event sent when a stream's title changes. - StreamTitleUpdated EventType = "STREAM_TITLE_UPDATED" - // SystemMessageSent is the event sent when a system message is sent. - SystemMessageSent EventType = "SYSTEM" - // ChatActionSent is a generic chat action that can be used for anything that doesn't need specific handling or formatting. - ChatActionSent EventType = "CHAT_ACTION" -) diff --git a/core/chat/events/fediverseEngagementEvent.go b/models/fediverseEngagementEvent.go index ceae96d16..611d59b39 100644 --- a/core/chat/events/fediverseEngagementEvent.go +++ b/models/fediverseEngagementEvent.go @@ -1,8 +1,4 @@ -package events - -import ( - "github.com/owncast/owncast/storage/configrepository" -) +package models // FediverseEngagementEvent is a message displayed in chat on representing an action on the Fediverse. type FediverseEngagementEvent struct { @@ -13,8 +9,6 @@ type FediverseEngagementEvent struct { UserAccountName string `json:"title"` } -var configRepository = configrepository.Get() - // GetBroadcastPayload will return the object to send to all chat users. func (e *FediverseEngagementEvent) GetBroadcastPayload() EventPayload { return EventPayload{ @@ -26,7 +20,7 @@ func (e *FediverseEngagementEvent) GetBroadcastPayload() EventPayload { "title": e.UserAccountName, "link": e.Link, "user": EventPayload{ - "displayName": configRepository.GetServerName(), + "displayName": "Owncast", }, } } diff --git a/core/chat/events/events.go b/models/messageEvents.go index 75083bf3d..741bee614 100644 --- a/core/chat/events/events.go +++ b/models/messageEvents.go @@ -1,4 +1,4 @@ -package events +package models import ( "bytes" @@ -9,7 +9,6 @@ import ( "time" "github.com/microcosm-cc/bluemonday" - "github.com/owncast/owncast/models" "github.com/teris-io/shortid" "github.com/yuin/goldmark" emoji "github.com/yuin/goldmark-emoji" @@ -42,9 +41,9 @@ type Event struct { // UserEvent is an event with an associated user. type UserEvent struct { - User *models.User `json:"user"` - HiddenAt *time.Time `json:"hiddenAt,omitempty"` - ClientID uint `json:"clientId,omitempty"` + User *User `json:"user"` + HiddenAt *time.Time `json:"hiddenAt,omitempty"` + ClientID uint `json:"clientId,omitempty"` } // MessageEvent is an event that has a message body. diff --git a/core/chat/events/nameChangeEvent.go b/models/nameChangeEvent.go index e5659a18f..de1e80cb6 100644 --- a/core/chat/events/nameChangeEvent.go +++ b/models/nameChangeEvent.go @@ -1,4 +1,4 @@ -package events +package models // NameChangeEvent is received when a user changes their chat display name. type NameChangeEvent struct { diff --git a/core/chat/events/setMessageVisibilityEvent.go b/models/setMessageVisibilityEvent.go index 6d82d6b0a..10a79ce62 100644 --- a/core/chat/events/setMessageVisibilityEvent.go +++ b/models/setMessageVisibilityEvent.go @@ -1,4 +1,4 @@ -package events +package models // SetMessageVisibilityEvent is the event fired when one or more message // visibilities are changed. diff --git a/models/stats.go b/models/stats.go index 6a69321d2..4bf7d18da 100644 --- a/models/stats.go +++ b/models/stats.go @@ -6,9 +6,9 @@ import ( // Stats holds the stats for the system. type Stats struct { + LastConnectTime *utils.NullTime `json:"lastConnectTime"` LastDisconnectTime *utils.NullTime `json:"lastDisconnectTime"` - LastConnectTime *utils.NullTime `json:"-"` ChatClients map[string]Client `json:"-"` Viewers map[string]*Viewer `json:"-"` SessionMaxViewerCount int `json:"sessionMaxViewerCount"` diff --git a/core/chat/events/systemMessageEvent.go b/models/systemMessageEvent.go index 9c9ec25df..ee9efd628 100644 --- a/core/chat/events/systemMessageEvent.go +++ b/models/systemMessageEvent.go @@ -1,9 +1,10 @@ -package events +package models // SystemMessageEvent is a message displayed in chat on behalf of the server. type SystemMessageEvent struct { Event MessageEvent + DisplayName string } // GetBroadcastPayload will return the object to send to all chat users. @@ -14,7 +15,7 @@ func (e *SystemMessageEvent) GetBroadcastPayload() EventPayload { "body": e.Body, "type": SystemMessageSent, "user": EventPayload{ - "displayName": configRepository.GetServerName(), + "displayName": e.DisplayName, }, } } diff --git a/core/chat/events/userDisabledEvent.go b/models/userDisabledEvent.go index d869210ef..53ad3bacb 100644 --- a/core/chat/events/userDisabledEvent.go +++ b/models/userDisabledEvent.go @@ -1,4 +1,4 @@ -package events +package models // UserDisabledEvent is the event fired when a user is banned/blocked and disconnected from chat. type UserDisabledEvent struct { diff --git a/models/userJoinedEvent.go b/models/userJoinedEvent.go index de7c55170..1d344d98c 100644 --- a/models/userJoinedEvent.go +++ b/models/userJoinedEvent.go @@ -1,11 +1,17 @@ package models -import "time" - -// UserJoinedEvent represents an event when a user joins the chat. +// UserJoinedEvent is the event fired when a user joins chat. type UserJoinedEvent struct { - Timestamp time.Time `json:"timestamp,omitempty"` - Username string `json:"username"` - Type EventType `json:"type"` - ID string `json:"id"` + Event + UserEvent +} + +// GetBroadcastPayload will return the object to send to all chat users. +func (e *UserJoinedEvent) GetBroadcastPayload() EventPayload { + return EventPayload{ + "type": UserJoined, + "id": e.ID, + "timestamp": e.Timestamp, + "user": e.User, + } } diff --git a/core/chat/events/userMessageEvent.go b/models/userMessageEvent.go index f0dd95e91..8961bcd5e 100644 --- a/core/chat/events/userMessageEvent.go +++ b/models/userMessageEvent.go @@ -1,4 +1,4 @@ -package events +package models // UserMessageEvent is an inbound message from a user. type UserMessageEvent struct { diff --git a/services/apfederation/activitypub.go b/services/apfederation/activitypub.go new file mode 100644 index 000000000..bad648943 --- /dev/null +++ b/services/apfederation/activitypub.go @@ -0,0 +1,82 @@ +package apfederation + +import ( + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/outbox" + + "github.com/owncast/owncast/services/apfederation/workerpool" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/data" + "github.com/owncast/owncast/storage/federationrepository" + + "github.com/owncast/owncast/models" + log "github.com/sirupsen/logrus" +) + +type APFederation struct { + workers *workerpool.WorkerPool + outbox *outbox.APOutbox +} + +func New() *APFederation { + ds := data.GetDatastore() + apf := &APFederation{ + outbox: outbox.Get(), + } + apf.Start(ds) + return apf +} + +var temporaryGlobalInstance *APFederation + +func Get() *APFederation { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} + +// Start will initialize and start the federation support. +func (ap *APFederation) Start(datastore *data.Store) { + configRepository := configrepository.Get() + + // workerpool.InitOutboundWorkerPool() + // ap.InitInboxWorkerPool() + + // Generate the keys for signing federated activity if needed. + if configRepository.GetPrivateKey() == "" { + privateKey, publicKey, err := crypto.GenerateKeys() + _ = configRepository.SetPrivateKey(string(privateKey)) + _ = configRepository.SetPublicKey(string(publicKey)) + if err != nil { + log.Errorln("Unable to get private key", err) + } + } +} + +// SendLive will send a "Go Live" message to followers. +func (ap *APFederation) SendLive() error { + return ap.SendLive() +} + +// SendPublicFederatedMessage will send an arbitrary provided message to followers. +func (ap *APFederation) SendPublicFederatedMessage(message string) error { + return ap.outbox.SendPublicMessage(message) +} + +// SendDirectFederatedMessage will send a direct message to a single account. +func (ap *APFederation) SendDirectFederatedMessage(message, account string) error { + return ap.outbox.SendDirectMessageToAccount(message, account) +} + +// GetFollowerCount will return the local tracked follower count. +func (ap *APFederation) GetFollowerCount() (int64, error) { + federationRepository := federationrepository.Get() + return federationRepository.GetFollowerCount() +} + +// GetPendingFollowRequests will return the pending follow requests. +func (ap *APFederation) GetPendingFollowRequests() ([]models.Follower, error) { + federationRepository := federationrepository.Get() + return federationRepository.GetPendingFollowRequests() +} diff --git a/activitypub/apmodels/activity.go b/services/apfederation/apmodels/activity.go index b31954e88..b31954e88 100644 --- a/activitypub/apmodels/activity.go +++ b/services/apfederation/apmodels/activity.go diff --git a/activitypub/apmodels/actor.go b/services/apfederation/apmodels/actor.go index becf15ab6..7ca9d00be 100644 --- a/activitypub/apmodels/actor.go +++ b/services/apfederation/apmodels/actor.go @@ -8,8 +8,8 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/crypto" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation/crypto" "github.com/owncast/owncast/storage/configrepository" log "github.com/sirupsen/logrus" ) diff --git a/activitypub/apmodels/actor_test.go b/services/apfederation/apmodels/actor_test.go index 94c5f9994..94c5f9994 100644 --- a/activitypub/apmodels/actor_test.go +++ b/services/apfederation/apmodels/actor_test.go diff --git a/activitypub/apmodels/hashtag.go b/services/apfederation/apmodels/hashtag.go index b7aca1d44..b7aca1d44 100644 --- a/activitypub/apmodels/hashtag.go +++ b/services/apfederation/apmodels/hashtag.go diff --git a/activitypub/apmodels/inboxRequest.go b/services/apfederation/apmodels/inboxRequest.go index 4a0c9f36b..4a0c9f36b 100644 --- a/activitypub/apmodels/inboxRequest.go +++ b/services/apfederation/apmodels/inboxRequest.go diff --git a/activitypub/apmodels/message.go b/services/apfederation/apmodels/message.go index 646057c36..646057c36 100644 --- a/activitypub/apmodels/message.go +++ b/services/apfederation/apmodels/message.go diff --git a/activitypub/apmodels/utils.go b/services/apfederation/apmodels/utils.go index 15b8c2065..15b8c2065 100644 --- a/activitypub/apmodels/utils.go +++ b/services/apfederation/apmodels/utils.go diff --git a/activitypub/apmodels/webfinger.go b/services/apfederation/apmodels/webfinger.go index f8151ca9d..f8151ca9d 100644 --- a/activitypub/apmodels/webfinger.go +++ b/services/apfederation/apmodels/webfinger.go diff --git a/activitypub/crypto/keys.go b/services/apfederation/crypto/keys.go index 2ee801a91..2ee801a91 100644 --- a/activitypub/crypto/keys.go +++ b/services/apfederation/crypto/keys.go diff --git a/activitypub/crypto/publicKey.go b/services/apfederation/crypto/publicKey.go index 5f2f524ad..5f2f524ad 100644 --- a/activitypub/crypto/publicKey.go +++ b/services/apfederation/crypto/publicKey.go diff --git a/activitypub/crypto/sign.go b/services/apfederation/crypto/sign.go index cc1cc481c..44154848b 100644 --- a/activitypub/crypto/sign.go +++ b/services/apfederation/crypto/sign.go @@ -77,7 +77,7 @@ func CreateSignedRequest(payload []byte, url *url.URL, fromActorIRI *url.URL) (* req, _ := http.NewRequest("POST", url.String(), bytes.NewBuffer(payload)) - c := config.GetConfig() + c := config.Get() ua := fmt.Sprintf("%s; https://owncast.online", c.GetReleaseString()) req.Header.Set("User-Agent", ua) diff --git a/activitypub/inbox/announce.go b/services/apfederation/inbox/announce.go index db79393f7..fe760458b 100644 --- a/activitypub/inbox/announce.go +++ b/services/apfederation/inbox/announce.go @@ -5,23 +5,22 @@ import ( "time" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" "github.com/pkg/errors" ) -func handleAnnounceRequest(c context.Context, activity vocab.ActivityStreamsAnnounce) error { +func (api *APInbox) handleAnnounceRequest(c context.Context, activity vocab.ActivityStreamsAnnounce) error { object := activity.GetActivityStreamsObject() actorReference := activity.GetActivityStreamsActor() objectIRI := object.At(0).GetIRI().String() actorIRI := actorReference.At(0).GetIRI().String() - if hasPreviouslyhandled, err := persistence.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, events.FediverseEngagementRepost); hasPreviouslyhandled || err != nil { + if hasPreviouslyhandled, err := api.federationRepository.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, models.FediverseEngagementRepost); hasPreviouslyhandled || err != nil { return errors.Wrap(err, "inbound activity of share/re-post has already been handled") } // Shares need to match a post we had already sent. - _, isLiveNotification, timestamp, err := persistence.GetObjectByIRI(objectIRI) + _, isLiveNotification, timestamp, err := api.federationRepository.GetObjectByIRI(objectIRI) if err != nil { return errors.Wrap(err, "Could not find post locally") } @@ -32,9 +31,9 @@ func handleAnnounceRequest(c context.Context, activity vocab.ActivityStreamsAnno } // Save as an accepted activity - if err := persistence.SaveInboundFediverseActivity(objectIRI, actorIRI, events.FediverseEngagementRepost, time.Now()); err != nil { + if err := api.federationRepository.SaveInboundFediverseActivity(objectIRI, actorIRI, models.FediverseEngagementRepost, time.Now()); err != nil { return errors.Wrap(err, "unable to save inbound share/re-post activity") } - return handleEngagementActivity(events.FediverseEngagementRepost, isLiveNotification, actorReference, events.FediverseEngagementRepost) + return api.handleEngagementActivity(models.FediverseEngagementRepost, isLiveNotification, actorReference, models.FediverseEngagementRepost) } diff --git a/services/apfederation/inbox/chat.go b/services/apfederation/inbox/chat.go new file mode 100644 index 000000000..a47a63ff5 --- /dev/null +++ b/services/apfederation/inbox/chat.go @@ -0,0 +1,59 @@ +package inbox + +import ( + "fmt" + + "github.com/go-fed/activity/streams/vocab" + "github.com/owncast/owncast/models" +) + +func (api *APInbox) handleEngagementActivity(eventType models.EventType, isLiveNotification bool, actorReference vocab.ActivityStreamsActorProperty, action string) error { + // Do nothing if displaying engagement actions has been turned off. + if !api.configRepository.GetFederationShowEngagement() { + return nil + } + + // Do nothing if chat is disabled + if api.configRepository.GetChatDisabled() { + return nil + } + + // Get actor of the action + actor, _ := api.resolvers.GetResolvedActorFromActorProperty(actorReference) + + // Send chat message + actorName := actor.Name + if actorName == "" { + actorName = actor.Username + } + actorIRI := actorReference.Begin().GetIRI().String() + + userPrefix := fmt.Sprintf("%s ", actorName) + var suffix string + if isLiveNotification && action == models.FediverseEngagementLike { + suffix = "liked that this stream went live." + } else if action == models.FediverseEngagementLike { + suffix = fmt.Sprintf("liked a post from %s.", api.configRepository.GetServerName()) + } else if isLiveNotification && action == models.FediverseEngagementRepost { + suffix = "shared this stream with their followers." + } else if action == models.FediverseEngagementRepost { + suffix = fmt.Sprintf("shared a post from %s.", api.configRepository.GetServerName()) + } else if action == models.FediverseEngagementFollow { + suffix = "followed this stream." + } else { + return fmt.Errorf("could not handle event for sending to chat: %s", action) + } + body := fmt.Sprintf("%s %s", userPrefix, suffix) + + var image *string + if actor.Image != nil { + s := actor.Image.String() + image = &s + } + + if err := api.chatService.SendFediverseAction(eventType, actor.FullUsername, image, body, actorIRI); err != nil { + return err + } + + return nil +} diff --git a/activitypub/inbox/constants.go b/services/apfederation/inbox/constants.go index a7dac5890..a7dac5890 100644 --- a/activitypub/inbox/constants.go +++ b/services/apfederation/inbox/constants.go diff --git a/activitypub/inbox/create.go b/services/apfederation/inbox/create.go index 3eab72a5e..a60f16569 100644 --- a/activitypub/inbox/create.go +++ b/services/apfederation/inbox/create.go @@ -7,7 +7,7 @@ import ( "github.com/pkg/errors" ) -func handleCreateRequest(c context.Context, activity vocab.ActivityStreamsCreate) error { +func (api *APInbox) handleCreateRequest(c context.Context, activity vocab.ActivityStreamsCreate) error { iri := activity.GetJSONLDId().GetIRI().String() return errors.New("not handling create request of: " + iri) } diff --git a/activitypub/inbox/follow.go b/services/apfederation/inbox/follow.go index e9e4ec0af..ed32115f9 100644 --- a/activitypub/inbox/follow.go +++ b/services/apfederation/inbox/follow.go @@ -6,17 +6,15 @@ import ( "time" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/requests" - "github.com/owncast/owncast/activitypub/resolvers" - "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation/outbox" "github.com/pkg/errors" log "github.com/sirupsen/logrus" ) -func handleFollowInboxRequest(c context.Context, activity vocab.ActivityStreamsFollow) error { - follow, err := resolvers.MakeFollowRequest(c, activity) +func (api *APInbox) handleFollowInboxRequest(c context.Context, activity vocab.ActivityStreamsFollow) error { + follow, err := api.resolvers.MakeFollowRequest(c, activity) if err != nil { log.Errorln("unable to create follow inbox request", err) return err @@ -26,19 +24,21 @@ func handleFollowInboxRequest(c context.Context, activity vocab.ActivityStreamsF return fmt.Errorf("unable to handle request") } - approved := !configRepository.GetFederationIsPrivate() + approved := !api.configRepository.GetFederationIsPrivate() followRequest := *follow - if err := persistence.AddFollow(followRequest, approved); err != nil { + if err := api.federationRepository.AddFollow(followRequest, approved); err != nil { log.Errorln("unable to save follow request", err) return err } - localAccountName := configRepository.GetDefaultFederationUsername() + localAccountName := api.configRepository.GetDefaultFederationUsername() + + ob := outbox.Get() if approved { - if err := requests.SendFollowAccept(follow.Inbox, activity, localAccountName); err != nil { + if err := ob.SendFollowAccept(follow.Inbox, activity, localAccountName); err != nil { log.Errorln("unable to send follow accept", err) return err } @@ -54,27 +54,27 @@ func handleFollowInboxRequest(c context.Context, activity vocab.ActivityStreamsF // chat due to a previous follow request, then do so. hasPreviouslyhandled := true // Default so we don't send anything if it fails. if approved { - hasPreviouslyhandled, err = persistence.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, events.FediverseEngagementFollow) + hasPreviouslyhandled, err = api.federationRepository.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, models.FediverseEngagementFollow) if err != nil { log.Errorln("error checking for previously handled follow activity", err) } } // Save this follow action to our activities table. - if err := persistence.SaveInboundFediverseActivity(objectIRI, actorIRI, events.FediverseEngagementFollow, time.Now()); err != nil { + if err := api.federationRepository.SaveInboundFediverseActivity(objectIRI, actorIRI, models.FediverseEngagementFollow, time.Now()); err != nil { return errors.Wrap(err, "unable to save inbound share/re-post activity") } // Send action to chat if it has not been previously handled. if !hasPreviouslyhandled { - return handleEngagementActivity(events.FediverseEngagementFollow, false, actorReference, events.FediverseEngagementFollow) + return api.handleEngagementActivity(models.FediverseEngagementFollow, false, actorReference, models.FediverseEngagementFollow) } return nil } -func handleUnfollowRequest(c context.Context, activity vocab.ActivityStreamsUndo) error { - request := resolvers.MakeUnFollowRequest(c, activity) +func (api *APInbox) handleUnfollowRequest(c context.Context, activity vocab.ActivityStreamsUndo) error { + request := api.resolvers.MakeUnFollowRequest(c, activity) if request == nil { log.Errorf("unable to handle unfollow request") return errors.New("unable to handle unfollow request") @@ -83,5 +83,5 @@ func handleUnfollowRequest(c context.Context, activity vocab.ActivityStreamsUndo unfollowRequest := *request log.Traceln("unfollow request:", unfollowRequest) - return persistence.RemoveFollow(unfollowRequest) + return api.federationRepository.RemoveFollow(unfollowRequest) } diff --git a/services/apfederation/inbox/inbox.go b/services/apfederation/inbox/inbox.go new file mode 100644 index 000000000..77d73d02b --- /dev/null +++ b/services/apfederation/inbox/inbox.go @@ -0,0 +1,37 @@ +package inbox + +import ( + "github.com/owncast/owncast/services/apfederation/requests" + "github.com/owncast/owncast/services/apfederation/resolvers" + "github.com/owncast/owncast/services/chat" + + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/federationrepository" +) + +type APInbox struct { + configRepository configrepository.ConfigRepository + federationRepository *federationrepository.FederationRepository + resolvers *resolvers.APResolvers + requests *requests.Requests + chatService *chat.Chat +} + +func New() *APInbox { + return &APInbox{ + configRepository: configrepository.Get(), + federationRepository: federationrepository.Get(), + resolvers: resolvers.Get(), + requests: requests.Get(), + chatService: chat.Get(), + } +} + +var temporaryGlobalInstance *APInbox + +func Get() *APInbox { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} diff --git a/activitypub/inbox/like.go b/services/apfederation/inbox/like.go index 4028d3d30..dd1313f06 100644 --- a/activitypub/inbox/like.go +++ b/services/apfederation/inbox/like.go @@ -5,12 +5,11 @@ import ( "time" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" "github.com/pkg/errors" ) -func handleLikeRequest(c context.Context, activity vocab.ActivityStreamsLike) error { +func (api *APInbox) handleLikeRequest(c context.Context, activity vocab.ActivityStreamsLike) error { object := activity.GetActivityStreamsObject() actorReference := activity.GetActivityStreamsActor() if object.Len() < 1 { @@ -24,12 +23,12 @@ func handleLikeRequest(c context.Context, activity vocab.ActivityStreamsLike) er objectIRI := object.At(0).GetIRI().String() actorIRI := actorReference.At(0).GetIRI().String() - if hasPreviouslyhandled, err := persistence.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, events.FediverseEngagementLike); hasPreviouslyhandled || err != nil { + if hasPreviouslyhandled, err := api.federationRepository.HasPreviouslyHandledInboundActivity(objectIRI, actorIRI, models.FediverseEngagementLike); hasPreviouslyhandled || err != nil { return errors.Wrap(err, "inbound activity of like has already been handled") } // Likes need to match a post we had already sent. - _, isLiveNotification, timestamp, err := persistence.GetObjectByIRI(objectIRI) + _, isLiveNotification, timestamp, err := api.federationRepository.GetObjectByIRI(objectIRI) if err != nil { return errors.Wrap(err, "Could not find post locally") } @@ -40,9 +39,9 @@ func handleLikeRequest(c context.Context, activity vocab.ActivityStreamsLike) er } // Save as an accepted activity - if err := persistence.SaveInboundFediverseActivity(objectIRI, actorIRI, events.FediverseEngagementLike, time.Now()); err != nil { + if err := api.federationRepository.SaveInboundFediverseActivity(objectIRI, actorIRI, models.FediverseEngagementLike, time.Now()); err != nil { return errors.Wrap(err, "unable to save inbound like activity") } - return handleEngagementActivity(events.FediverseEngagementLike, isLiveNotification, actorReference, events.FediverseEngagementLike) + return api.handleEngagementActivity(models.FediverseEngagementLike, isLiveNotification, actorReference, models.FediverseEngagementLike) } diff --git a/activitypub/inbox/undo.go b/services/apfederation/inbox/undo.go index fd18fc6bb..9cbf166f0 100644 --- a/activitypub/inbox/undo.go +++ b/services/apfederation/inbox/undo.go @@ -8,13 +8,13 @@ import ( "github.com/go-fed/activity/streams/vocab" ) -func handleUndoInboxRequest(c context.Context, activity vocab.ActivityStreamsUndo) error { +func (api *APInbox) handleUndoInboxRequest(c context.Context, activity vocab.ActivityStreamsUndo) error { // Determine if this is an undo of a follow, favorite, announce, etc. o := activity.GetActivityStreamsObject() for iter := o.Begin(); iter != o.End(); iter = iter.Next() { if iter.IsActivityStreamsFollow() { // This is an Unfollow request - if err := handleUnfollowRequest(c, activity); err != nil { + if err := api.handleUnfollowRequest(c, activity); err != nil { return err } } else { diff --git a/services/apfederation/inbox/update.go b/services/apfederation/inbox/update.go new file mode 100644 index 000000000..879f07fe8 --- /dev/null +++ b/services/apfederation/inbox/update.go @@ -0,0 +1,23 @@ +package inbox + +import ( + "context" + + "github.com/go-fed/activity/streams/vocab" + log "github.com/sirupsen/logrus" +) + +func (api *APInbox) handleUpdateRequest(c context.Context, activity vocab.ActivityStreamsUpdate) error { + // We only care about update events to followers. + if !activity.GetActivityStreamsObject().At(0).IsActivityStreamsPerson() { + return nil + } + + actor, err := api.resolvers.GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) + if err != nil { + log.Errorln(err) + return err + } + + return api.federationRepository.UpdateFollower(actor.ActorIri.String(), actor.Inbox.String(), actor.Name, actor.FullUsername, actor.Image.String()) +} diff --git a/activitypub/inbox/worker.go b/services/apfederation/inbox/worker.go index 008d6d5a7..9695a5bf7 100644 --- a/activitypub/inbox/worker.go +++ b/services/apfederation/inbox/worker.go @@ -12,15 +12,13 @@ import ( "github.com/pkg/errors" "github.com/go-fed/httpsig" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/resolvers" + "github.com/owncast/owncast/services/apfederation/apmodels" log "github.com/sirupsen/logrus" ) -func handle(request apmodels.InboxRequest) { - if verified, err := Verify(request.Request); err != nil { +func (api *APInbox) handle(request apmodels.InboxRequest) { + if verified, err := api.Verify(request.Request); err != nil { log.Debugln("Error in attempting to verify request", err) return } else if !verified { @@ -28,7 +26,7 @@ func handle(request apmodels.InboxRequest) { return } - if err := resolvers.Resolve(context.Background(), request.Body, handleUpdateRequest, handleFollowInboxRequest, handleLikeRequest, handleAnnounceRequest, handleUndoInboxRequest, handleCreateRequest); err != nil { + if err := api.resolvers.Resolve(context.Background(), request.Body, api.handleUpdateRequest, api.handleFollowInboxRequest, api.handleLikeRequest, api.handleAnnounceRequest, api.handleUndoInboxRequest, api.handleCreateRequest); err != nil { log.Debugln("resolver error:", err) } } @@ -36,7 +34,7 @@ func handle(request apmodels.InboxRequest) { // Verify will Verify the http signature of an inbound request as well as // check it against the list of blocked domains. // nolint: cyclop -func Verify(request *http.Request) (bool, error) { +func (api *APInbox) Verify(request *http.Request) (bool, error) { verifier, err := httpsig.NewVerifier(request) if err != nil { return false, errors.Wrap(err, "failed to create key verifier for request") @@ -71,7 +69,7 @@ func Verify(request *http.Request) (bool, error) { return false, errors.New("Unable to determine algorithm to verify request") } - publicKey, err := resolvers.GetResolvedPublicKeyFromIRI(pubKeyID.String()) + publicKey, err := api.resolvers.GetResolvedPublicKeyFromIRI(pubKeyID.String()) if err != nil { return false, errors.Wrap(err, "failed to resolve actor from IRI to fetch key") } @@ -86,12 +84,12 @@ func Verify(request *http.Request) (bool, error) { } // Test to see if the actor is in the list of blocked federated domains. - if isBlockedDomain(publicKeyActorIRI.Hostname()) { + if api.isBlockedDomain(publicKeyActorIRI.Hostname()) { return false, errors.New("domain is blocked") } // If actor is specifically blocked, then fail validation. - if blocked, err := isBlockedActor(publicKeyActorIRI); err != nil || blocked { + if blocked, err := api.isBlockedActor(publicKeyActorIRI); err != nil || blocked { return false, err } @@ -129,8 +127,8 @@ func Verify(request *http.Request) (bool, error) { return false, fmt.Errorf("http signature verification error(s) for: %s: %+v", pubKeyID.String(), triedAlgos) } -func isBlockedDomain(domain string) bool { - blockedDomains := configRepository.GetBlockedFederatedDomains() +func (api *APInbox) isBlockedDomain(domain string) bool { + blockedDomains := api.configRepository.GetBlockedFederatedDomains() for _, blockedDomain := range blockedDomains { if strings.Contains(domain, blockedDomain) { @@ -141,8 +139,8 @@ func isBlockedDomain(domain string) bool { return false } -func isBlockedActor(actorIRI *url.URL) (bool, error) { - blockedactor, err := persistence.GetFollower(actorIRI.String()) +func (api *APInbox) isBlockedActor(actorIRI *url.URL) (bool, error) { + blockedactor, err := api.federationRepository.GetFollower(actorIRI.String()) if blockedactor != nil && blockedactor.DisabledAt != nil { return true, errors.Wrap(err, "remote actor is blocked") diff --git a/activitypub/inbox/worker_test.go b/services/apfederation/inbox/worker_test.go index 2b4b8d956..98dec41f3 100644 --- a/activitypub/inbox/worker_test.go +++ b/services/apfederation/inbox/worker_test.go @@ -6,10 +6,10 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/persistence" + "github.com/owncast/owncast/services/apfederation/apmodels" "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/storage/data" + "github.com/owncast/owncast/storage/federationrepository" ) func makeFakePerson() vocab.ActivityStreamsPerson { @@ -54,6 +54,7 @@ func TestMain(m *testing.M) { panic(err) } + _ = federationrepository.New(ds) configRepository := configrepository.New(ds) configRepository.PopulateDefaults() configRepository.SetServerURL("https://my.cool.site.biz") @@ -82,17 +83,20 @@ func TestBlockedDomains(t *testing.T) { } func TestBlockedActors(t *testing.T) { + federationRepository := federationrepository.Get() person := makeFakePerson() fakeRequest := streams.NewActivityStreamsFollow() - persistence.AddFollow(apmodels.ActivityPubActor{ + ib := Get() + + federationRepository.AddFollow(apmodels.ActivityPubActor{ ActorIri: person.GetJSONLDId().GetIRI(), Inbox: person.GetJSONLDId().GetIRI(), FollowRequestIri: person.GetJSONLDId().GetIRI(), RequestObject: fakeRequest, }, false) - persistence.BlockOrRejectFollower(person.GetJSONLDId().GetIRI().String()) + federationRepository.BlockOrRejectFollower(person.GetJSONLDId().GetIRI().String()) - blocked, err := isBlockedActor(person.GetJSONLDId().GetIRI()) + blocked, err := ib.isBlockedActor(person.GetJSONLDId().GetIRI()) if err != nil { t.Error(err) return @@ -103,7 +107,7 @@ func TestBlockedActors(t *testing.T) { } failedBlockIRI, _ := url.Parse("https://freedom.eagle/user/mrbar") - failedBlock, err := isBlockedActor(failedBlockIRI) + failedBlock, err := ib.isBlockedActor(failedBlockIRI) if failedBlock { t.Error("Invalid blocking of unblocked actor IRI") diff --git a/activitypub/inbox/workerpool.go b/services/apfederation/inbox/workerpool.go index 38bc2100e..fcf9de120 100644 --- a/activitypub/inbox/workerpool.go +++ b/services/apfederation/inbox/workerpool.go @@ -4,6 +4,7 @@ import ( "runtime" "github.com/owncast/owncast/activitypub/apmodels" + "github.com/owncast/owncast/services/apfederation/apmodels" log "github.com/sirupsen/logrus" ) @@ -18,7 +19,7 @@ type Job struct { var queue chan Job // InitInboxWorkerPool starts n go routines that await ActivityPub jobs. -func InitInboxWorkerPool() { +func (api *APInbox) InitInboxWorkerPool() { queue = make(chan Job) // start workers @@ -28,16 +29,16 @@ func InitInboxWorkerPool() { } // AddToQueue will queue up an outbound http request. -func AddToQueue(req apmodels.InboxRequest) { +func (api *APInbox) AddToQueue(req apmodels.InboxRequest) { log.Tracef("Queued request for ActivityPub inbox handler") queue <- Job{req} } -func worker(workerID int, queue <-chan Job) { +func (api *APInbox) worker(workerID int, queue <-chan Job) { log.Debugf("Started ActivityPub worker %d", workerID) for job := range queue { - handle(job.request) + api.handle(job.request) log.Tracef("Done with ActivityPub inbox handler using worker %d", workerID) } diff --git a/activitypub/requests/acceptFollow.go b/services/apfederation/outbox/acceptFollow.go index add89e24f..1c0d9c8e5 100644 --- a/activitypub/requests/acceptFollow.go +++ b/services/apfederation/outbox/acceptFollow.go @@ -1,4 +1,4 @@ -package requests +package outbox import ( "encoding/json" @@ -6,16 +6,15 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/workerpool" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" "github.com/teris-io/shortid" ) // SendFollowAccept will send an accept activity to a follow request from a specified local user. -func SendFollowAccept(inbox *url.URL, originalFollowActivity vocab.ActivityStreamsFollow, fromLocalAccountName string) error { - followAccept := makeAcceptFollow(originalFollowActivity, fromLocalAccountName) +func (apo *APOutbox) SendFollowAccept(inbox *url.URL, originalFollowActivity vocab.ActivityStreamsFollow, fromLocalAccountName string) error { + followAccept := apo.makeAcceptFollow(originalFollowActivity, fromLocalAccountName) localAccountIRI := apmodels.MakeLocalIRIForAccount(fromLocalAccountName) var jsonmap map[string]interface{} @@ -26,12 +25,12 @@ func SendFollowAccept(inbox *url.URL, originalFollowActivity vocab.ActivityStrea return err } - workerpool.AddToOutboundQueue(req) + apo.workerpool.AddToOutboundQueue(req) return nil } -func makeAcceptFollow(originalFollowActivity vocab.ActivityStreamsFollow, fromAccountName string) vocab.ActivityStreamsAccept { +func (r *APOutbox) makeAcceptFollow(originalFollowActivity vocab.ActivityStreamsFollow, fromAccountName string) vocab.ActivityStreamsAccept { acceptIDString := shortid.MustGenerate() acceptID := apmodels.MakeLocalIRIForResource(acceptIDString) actorID := apmodels.MakeLocalIRIForAccount(fromAccountName) diff --git a/activitypub/outbox/outbox.go b/services/apfederation/outbox/outbox.go index eeced48e1..0170d731b 100644 --- a/activitypub/outbox/outbox.go +++ b/services/apfederation/outbox/outbox.go @@ -9,6 +9,7 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" +<<<<<<< HEAD:activitypub/outbox/outbox.go "github.com/owncast/owncast/activitypub/apmodels" "github.com/owncast/owncast/activitypub/crypto" "github.com/owncast/owncast/activitypub/persistence" @@ -17,20 +18,55 @@ import ( "github.com/owncast/owncast/activitypub/webfinger" "github.com/owncast/owncast/activitypub/workerpool" "github.com/owncast/owncast/core/data" +======= + +>>>>>>> 4f9fbfba1 (WIP):services/apfederation/outbox/outbox.go "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/federationrepository" "github.com/pkg/errors" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" + "github.com/owncast/owncast/services/apfederation/resolvers" + "github.com/owncast/owncast/services/apfederation/webfinger" + "github.com/owncast/owncast/services/apfederation/workerpool" "github.com/owncast/owncast/services/config" "github.com/owncast/owncast/utils" log "github.com/sirupsen/logrus" "github.com/teris-io/shortid" ) -var configRepository = configrepository.Get() +type APOutbox struct { + configRepository configrepository.ConfigRepository + federationRepository *federationrepository.FederationRepository + resolvers *resolvers.APResolvers + workerpool *workerpool.WorkerPool + requests *requests.Requests +} + +func New() *APOutbox { + return &APOutbox{ + configRepository: configrepository.Get(), + federationRepository: federationrepository.Get(), + resolvers: resolvers.Get(), + workerpool: workerpool.Get(), + requests: requests.Get(), + } +} + +var temporaryGlobalInstance *APOutbox + +func Get() *APOutbox { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} // SendLive will send all followers the message saying you started a live stream. -func SendLive() error { - textContent := configRepository.GetFederationGoLiveMessage() +func (apo *APOutbox) SendLive() error { + textContent := apo.configRepository.GetFederationGoLiveMessage() // If the message is empty then do not send it. if textContent == "" { @@ -41,11 +77,11 @@ func SendLive() error { reg := regexp.MustCompile("[^a-zA-Z0-9]+") tagProp := streams.NewActivityStreamsTagProperty() - for _, tagString := range configRepository.GetServerMetadataTags() { + for _, tagString := range apo.configRepository.GetServerMetadataTags() { tagWithoutSpecialCharacters := reg.ReplaceAllString(tagString, "") hashtag := apmodels.MakeHashtag(tagWithoutSpecialCharacters) tagProp.AppendTootHashtag(hashtag) - tagString := getHashtagLinkHTMLFromTagString(tagWithoutSpecialCharacters) + tagString := apo.getHashtagLinkHTMLFromTagString(tagWithoutSpecialCharacters) tagStrings = append(tagStrings, tagString) } @@ -60,15 +96,19 @@ func SendLive() error { tagsString := strings.Join(tagStrings, " ") var streamTitle string - if title := configRepository.GetStreamTitle(); title != "" { + if title := apo.configRepository.GetStreamTitle(); title != "" { streamTitle = fmt.Sprintf("<p>%s</p>", title) } +<<<<<<< HEAD:activitypub/outbox/outbox.go textContent = fmt.Sprintf("<p>%s</p>%s<p>%s</p><p><a href=\"%s\">%s</a></p>", textContent, streamTitle, tagsString, data.GetServerURL(), data.GetServerURL()) +======= + textContent = fmt.Sprintf("<p>%s</p>%s<p>%s</p><a href=\"%s\">%s</a>", textContent, streamTitle, tagsString, apo.configRepository.GetServerURL(), apo.configRepository.GetServerURL()) +>>>>>>> 4f9fbfba1 (WIP):services/apfederation/outbox/outbox.go - activity, _, note, noteID := createBaseOutboundMessage(textContent) + activity, _, note, noteID := apo.createBaseOutboundMessage(textContent) // To the public if we're not treating ActivityPub as "private". - if !configRepository.GetFederationIsPrivate() { + if !apo.configRepository.GetFederationIsPrivate() { note = apmodels.MakeNotePublic(note) activity = apmodels.MakeActivityPublic(activity) } @@ -76,11 +116,11 @@ func SendLive() error { note.SetActivityStreamsTag(tagProp) // Attach an image along with the Federated message. - previewURL, err := url.Parse(configRepository.GetServerURL()) + previewURL, err := url.Parse(apo.configRepository.GetServerURL()) if err == nil { var imageToAttach string var mediaType string - c := config.GetConfig() + c := config.Get() previewGif := filepath.Join(c.TempDir, "preview.gif") thumbnailJpg := filepath.Join(c.TempDir, "thumbnail.jpg") uniquenessString := shortid.MustGenerate() @@ -98,7 +138,7 @@ func SendLive() error { } } - if configRepository.GetNSFW() { + if apo.configRepository.GetNSFW() { // Mark content as sensitive. sensitive := streams.NewActivityStreamsSensitiveProperty() sensitive.AppendXMLSchemaBoolean(true) @@ -111,11 +151,11 @@ func SendLive() error { return errors.New("unable to serialize go live message activity " + err.Error()) } - if err := SendToFollowers(b); err != nil { + if err := apo.SendToFollowers(b); err != nil { return err } - if err := Add(note, noteID, true); err != nil { + if err := apo.Add(note, noteID, true); err != nil { return err } @@ -123,20 +163,22 @@ func SendLive() error { } // SendDirectMessageToAccount will send a direct message to a single account. -func SendDirectMessageToAccount(textContent, account string) error { - links, err := webfinger.GetWebfingerLinks(account) +func (apo *APOutbox) SendDirectMessageToAccount(textContent, account string) error { + wf := webfinger.Get() + + links, err := wf.GetWebfingerLinks(account) if err != nil { return errors.Wrap(err, "unable to get webfinger links when sending private message") } user := apmodels.MakeWebFingerRequestResponseFromData(links) iri := user.Self - actor, err := resolvers.GetResolvedActorFromIRI(iri) + actor, err := apo.resolvers.GetResolvedActorFromIRI(iri) if err != nil { return errors.Wrap(err, "unable to resolve actor to send message to") } - activity, _, note, _ := createBaseOutboundMessage(textContent) + activity, _, note, _ := apo.createBaseOutboundMessage(textContent) // Set direct message visibility activity = apmodels.MakeActivityDirect(activity, actor.ActorIri) @@ -150,11 +192,11 @@ func SendDirectMessageToAccount(textContent, account string) error { return errors.Wrap(err, "unable to serialize custom fediverse message activity") } - return SendToUser(actor.Inbox, b) + return apo.SendToUser(actor.Inbox, b) } // SendPublicMessage will send a public message to all followers. -func SendPublicMessage(textContent string) error { +func (apo *APOutbox) SendPublicMessage(textContent string) error { originalContent := textContent textContent = utils.RenderSimpleMarkdown(textContent) @@ -166,7 +208,7 @@ func SendPublicMessage(textContent string) error { tagWithoutHashtag := strings.TrimPrefix(hashtag, "#") // Replace the instances of the tag with a link to the tag page. - tagHTML := getHashtagLinkHTMLFromTagString(tagWithoutHashtag) + tagHTML := apo.getHashtagLinkHTMLFromTagString(tagWithoutHashtag) textContent = strings.ReplaceAll(textContent, hashtag, tagHTML) // Create Hashtag object for the tag. @@ -174,10 +216,10 @@ func SendPublicMessage(textContent string) error { tagProp.AppendTootHashtag(hashtag) } - activity, _, note, noteID := createBaseOutboundMessage(textContent) + activity, _, note, noteID := apo.createBaseOutboundMessage(textContent) note.SetActivityStreamsTag(tagProp) - if !configRepository.GetFederationIsPrivate() { + if !apo.configRepository.GetFederationIsPrivate() { note = apmodels.MakeNotePublic(note) activity = apmodels.MakeActivityPublic(activity) } @@ -188,11 +230,11 @@ func SendPublicMessage(textContent string) error { return errors.New("unable to serialize custom fediverse message activity " + err.Error()) } - if err := SendToFollowers(b); err != nil { + if err := apo.SendToFollowers(b); err != nil { return err } - if err := Add(note, noteID, false); err != nil { + if err := apo.Add(note, noteID, false); err != nil { return err } @@ -200,8 +242,8 @@ func SendPublicMessage(textContent string) error { } // nolint: unparam -func createBaseOutboundMessage(textContent string) (vocab.ActivityStreamsCreate, string, vocab.ActivityStreamsNote, string) { - localActor := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername()) +func (apo *APOutbox) createBaseOutboundMessage(textContent string) (vocab.ActivityStreamsCreate, string, vocab.ActivityStreamsNote, string) { + localActor := apmodels.MakeLocalIRIForAccount(apo.configRepository.GetDefaultFederationUsername()) noteID := shortid.MustGenerate() noteIRI := apmodels.MakeLocalIRIForResource(noteID) id := shortid.MustGenerate() @@ -216,15 +258,15 @@ func createBaseOutboundMessage(textContent string) (vocab.ActivityStreamsCreate, } // Get Hashtag HTML link for a given tag (without # prefix). -func getHashtagLinkHTMLFromTagString(baseHashtag string) string { +func (apo *APOutbox) getHashtagLinkHTMLFromTagString(baseHashtag string) string { return fmt.Sprintf("<a class=\"hashtag\" href=\"https://directory.owncast.online/tags/%s\">#%s</a>", baseHashtag, baseHashtag) } // SendToFollowers will send an arbitrary payload to all follower inboxes. -func SendToFollowers(payload []byte) error { - localActor := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername()) +func (apo *APOutbox) SendToFollowers(payload []byte) error { + localActor := apmodels.MakeLocalIRIForAccount(apo.configRepository.GetDefaultFederationUsername()) - followers, _, err := persistence.GetFederationFollowers(-1, 0) + followers, _, err := apo.federationRepository.GetFederationFollowers(-1, 0) if err != nil { log.Errorln("unable to fetch followers to send to", err) return errors.New("unable to fetch followers to send payload to") @@ -238,29 +280,29 @@ func SendToFollowers(payload []byte) error { return errors.New("unable to create outbox request: " + follower.Inbox) } - workerpool.AddToOutboundQueue(req) + apo.workerpool.AddToOutboundQueue(req) } return nil } // SendToUser will send a payload to a single specific inbox. -func SendToUser(inbox *url.URL, payload []byte) error { - localActor := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername()) +func (apo *APOutbox) SendToUser(inbox *url.URL, payload []byte) error { + localActor := apmodels.MakeLocalIRIForAccount(apo.configRepository.GetDefaultFederationUsername()) - req, err := requests.CreateSignedRequest(payload, inbox, localActor) + req, err := apo.requests.CreateSignedRequest(payload, inbox, localActor) if err != nil { return errors.Wrap(err, "unable to create outbox request") } - workerpool.AddToOutboundQueue(req) + apo.workerpool.AddToOutboundQueue(req) return nil } // UpdateFollowersWithAccountUpdates will send an update to all followers alerting of a profile update. -func UpdateFollowersWithAccountUpdates() error { +func (apo *APOutbox) UpdateFollowersWithAccountUpdates() error { // Don't do anything if federation is disabled. - if !configRepository.GetFederationEnabled() { + if !apo.configRepository.GetFederationEnabled() { return nil } @@ -269,7 +311,7 @@ func UpdateFollowersWithAccountUpdates() error { activity := apmodels.MakeUpdateActivity(objectID) actor := streams.NewActivityStreamsPerson() - actorID := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername()) + actorID := apmodels.MakeLocalIRIForAccount(apo.configRepository.GetDefaultFederationUsername()) actorIDProperty := streams.NewJSONLDIdProperty() actorIDProperty.Set(actorID) actor.SetJSONLDId(actorIDProperty) @@ -287,11 +329,11 @@ func UpdateFollowersWithAccountUpdates() error { log.Errorln("unable to serialize send update actor activity", err) return errors.New("unable to serialize send update actor activity") } - return SendToFollowers(b) + return apo.SendToFollowers(b) } // Add will save an ActivityPub object to the datastore. -func Add(item vocab.Type, id string, isLiveNotification bool) error { +func (apo *APOutbox) Add(item vocab.Type, id string, isLiveNotification bool) error { iri := item.GetJSONLDId().GetIRI().String() typeString := item.GetTypeName() @@ -306,5 +348,5 @@ func Add(item vocab.Type, id string, isLiveNotification bool) error { return err } - return persistence.AddToOutbox(iri, b, typeString, isLiveNotification) + return apo.federationRepository.AddToOutbox(iri, b, typeString, isLiveNotification) } diff --git a/activitypub/requests/http.go b/services/apfederation/requests/http.go index c23c0eeac..360468b6c 100644 --- a/activitypub/requests/http.go +++ b/services/apfederation/requests/http.go @@ -9,14 +9,14 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/crypto" + "github.com/owncast/owncast/services/apfederation/crypto" "github.com/owncast/owncast/services/config" log "github.com/sirupsen/logrus" ) // WriteStreamResponse will write a ActivityPub object to the provided ResponseWriter and sign with the provided key. -func WriteStreamResponse(item vocab.Type, w http.ResponseWriter, publicKey crypto.PublicKey) error { +func (r *Requests) WriteStreamResponse(item vocab.Type, w http.ResponseWriter, publicKey crypto.PublicKey) error { var jsonmap map[string]interface{} jsonmap, _ = streams.Serialize(item) b, err := json.Marshal(jsonmap) @@ -24,21 +24,21 @@ func WriteStreamResponse(item vocab.Type, w http.ResponseWriter, publicKey crypt return err } - return WriteResponse(b, w, publicKey) + return r.WriteResponse(b, w, publicKey) } // WritePayloadResponse will write any arbitrary object to the provided ResponseWriter and sign with the provided key. -func WritePayloadResponse(payload interface{}, w http.ResponseWriter, publicKey crypto.PublicKey) error { +func (r *Requests) WritePayloadResponse(payload interface{}, w http.ResponseWriter, publicKey crypto.PublicKey) error { b, err := json.Marshal(payload) if err != nil { return err } - return WriteResponse(b, w, publicKey) + return r.WriteResponse(b, w, publicKey) } // WriteResponse will write any arbitrary payload to the provided ResponseWriter and sign with the provided key. -func WriteResponse(payload []byte, w http.ResponseWriter, publicKey crypto.PublicKey) error { +func (r *Requests) WriteResponse(payload []byte, w http.ResponseWriter, publicKey crypto.PublicKey) error { w.Header().Set("Content-Type", "application/activity+json") if err := crypto.SignResponse(w, payload, publicKey); err != nil { @@ -56,11 +56,11 @@ func WriteResponse(payload []byte, w http.ResponseWriter, publicKey crypto.Publi } // CreateSignedRequest will create a signed POST request of a payload to the provided destination. -func CreateSignedRequest(payload []byte, url *url.URL, fromActorIRI *url.URL) (*http.Request, error) { +func (r *Requests) CreateSignedRequest(payload []byte, url *url.URL, fromActorIRI *url.URL) (*http.Request, error) { log.Debugln("Sending", string(payload), "to", url) req, _ := http.NewRequest(http.MethodPost, url.String(), bytes.NewBuffer(payload)) - c := config.GetConfig() + c := config.Get() ua := fmt.Sprintf("%s; https://owncast.online", c.GetReleaseString()) req.Header.Set("User-Agent", ua) req.Header.Set("Content-Type", "application/activity+json") diff --git a/services/apfederation/requests/requests.go b/services/apfederation/requests/requests.go new file mode 100644 index 000000000..c2769918e --- /dev/null +++ b/services/apfederation/requests/requests.go @@ -0,0 +1,20 @@ +package requests + +import "github.com/owncast/owncast/services/apfederation/workerpool" + +type Requests struct { + outboundWorkerPool *workerpool.WorkerPool +} + +func New() *Requests { + return &Requests{} +} + +var temporaryGlobalInstance *Requests + +func Get() *Requests { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} diff --git a/activitypub/resolvers/follow.go b/services/apfederation/resolvers/follow.go index 42211c9cd..c61b2a327 100644 --- a/activitypub/resolvers/follow.go +++ b/services/apfederation/resolvers/follow.go @@ -5,18 +5,18 @@ import ( "fmt" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" + "github.com/owncast/owncast/services/apfederation/apmodels" "github.com/pkg/errors" log "github.com/sirupsen/logrus" ) -func getPersonFromFollow(activity vocab.ActivityStreamsFollow) (apmodels.ActivityPubActor, error) { - return GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) +func (apr *APResolvers) getPersonFromFollow(activity vocab.ActivityStreamsFollow) (apmodels.ActivityPubActor, error) { + return apr.GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) } // MakeFollowRequest will convert an inbound Follow request to our internal actor model. -func MakeFollowRequest(c context.Context, activity vocab.ActivityStreamsFollow) (*apmodels.ActivityPubActor, error) { - person, err := getPersonFromFollow(activity) +func (apr *APResolvers) MakeFollowRequest(c context.Context, activity vocab.ActivityStreamsFollow) (*apmodels.ActivityPubActor, error) { + person, err := apr.getPersonFromFollow(activity) if err != nil { return nil, errors.New("unable to resolve person from follow request: " + err.Error()) } @@ -39,8 +39,8 @@ func MakeFollowRequest(c context.Context, activity vocab.ActivityStreamsFollow) } // MakeUnFollowRequest will convert an inbound Unfollow request to our internal actor model. -func MakeUnFollowRequest(c context.Context, activity vocab.ActivityStreamsUndo) *apmodels.ActivityPubActor { - person, err := GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) +func (apr *APResolvers) MakeUnFollowRequest(c context.Context, activity vocab.ActivityStreamsUndo) *apmodels.ActivityPubActor { + person, err := apr.GetResolvedActorFromActorProperty(activity.GetActivityStreamsActor()) if err != nil { log.Errorln("unable to resolve person from actor iri", person.ActorIri, err) return nil diff --git a/activitypub/resolvers/resolve.go b/services/apfederation/resolvers/resolve.go index 73be0405b..444068642 100644 --- a/activitypub/resolvers/resolve.go +++ b/services/apfederation/resolvers/resolve.go @@ -8,15 +8,15 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" "github.com/owncast/owncast/storage/configrepository" "github.com/pkg/errors" log "github.com/sirupsen/logrus" ) // Resolve will translate a raw ActivityPub payload and fire the callback associated with that activity type. -func Resolve(c context.Context, data []byte, callbacks ...interface{}) error { +func (apr *APResolvers) Resolve(c context.Context, data []byte, callbacks ...interface{}) error { jsonResolver, err := streams.NewJSONResolver(callbacks...) if err != nil { // Something in the setup was wrong. For example, a callback has an @@ -46,7 +46,7 @@ func Resolve(c context.Context, data []byte, callbacks ...interface{}) error { } // ResolveIRI will resolve an IRI ahd call the correct callback for the resolved type. -func ResolveIRI(c context.Context, iri string, callbacks ...interface{}) error { +func (apr *APResolvers) ResolveIRI(c context.Context, iri string, callbacks ...interface{}) error { log.Debugln("Resolving", iri) req, _ := http.NewRequest(http.MethodGet, iri, nil) @@ -71,12 +71,12 @@ func ResolveIRI(c context.Context, iri string, callbacks ...interface{}) error { } // fmt.Println(string(data)) - return Resolve(c, data, callbacks...) + return apr.Resolve(c, data, callbacks...) } // GetResolvedActorFromActorProperty resolve an external actor property to a // fully populated internal actor representation. -func GetResolvedActorFromActorProperty(actor vocab.ActivityStreamsActorProperty) (apmodels.ActivityPubActor, error) { +func (apr *APResolvers) GetResolvedActorFromActorProperty(actor vocab.ActivityStreamsActorProperty) (apmodels.ActivityPubActor, error) { var err error var apActor apmodels.ActivityPubActor resolved := false @@ -89,7 +89,7 @@ func GetResolvedActorFromActorProperty(actor vocab.ActivityStreamsActorProperty) // If the actor is an unresolved IRI then we need to resolve it. if actorObjectOrIRI.IsIRI() { iri := actorObjectOrIRI.GetIRI().String() - return GetResolvedActorFromIRI(iri) + return apr.GetResolvedActorFromIRI(iri) } if actorObjectOrIRI.IsActivityStreamsPerson() { @@ -125,7 +125,7 @@ func GetResolvedActorFromActorProperty(actor vocab.ActivityStreamsActorProperty) } // GetResolvedPublicKeyFromIRI will resolve a publicKey IRI string to a vocab.W3IDSecurityV1PublicKey. -func GetResolvedPublicKeyFromIRI(publicKeyIRI string) (vocab.W3IDSecurityV1PublicKey, error) { +func (apr *APResolvers) GetResolvedPublicKeyFromIRI(publicKeyIRI string) (vocab.W3IDSecurityV1PublicKey, error) { var err error var pubkey vocab.W3IDSecurityV1PublicKey resolved := false @@ -175,7 +175,7 @@ func GetResolvedPublicKeyFromIRI(publicKeyIRI string) (vocab.W3IDSecurityV1Publi return nil } - if e := ResolveIRI(context.Background(), publicKeyIRI, personCallback, serviceCallback, applicationCallback, pubkeyCallback); e != nil { + if e := apr.ResolveIRI(context.Background(), publicKeyIRI, personCallback, serviceCallback, applicationCallback, pubkeyCallback); e != nil { err = e } @@ -191,7 +191,7 @@ func GetResolvedPublicKeyFromIRI(publicKeyIRI string) (vocab.W3IDSecurityV1Publi } // GetResolvedActorFromIRI will resolve an IRI string to a fully populated actor. -func GetResolvedActorFromIRI(personOrServiceIRI string) (apmodels.ActivityPubActor, error) { +func (apr *APResolvers) GetResolvedActorFromIRI(personOrServiceIRI string) (apmodels.ActivityPubActor, error) { var err error var apActor apmodels.ActivityPubActor resolved := false @@ -222,7 +222,7 @@ func GetResolvedActorFromIRI(personOrServiceIRI string) (apmodels.ActivityPubAct return e } - if e := ResolveIRI(context.Background(), personOrServiceIRI, personCallback, serviceCallback, applicationCallback); e != nil { + if e := apr.ResolveIRI(context.Background(), personOrServiceIRI, personCallback, serviceCallback, applicationCallback); e != nil { err = e } diff --git a/services/apfederation/resolvers/resolvers.go b/services/apfederation/resolvers/resolvers.go new file mode 100644 index 000000000..caeb4b19b --- /dev/null +++ b/services/apfederation/resolvers/resolvers.go @@ -0,0 +1,24 @@ +package resolvers + +import ( + "github.com/owncast/owncast/storage/configrepository" +) + +type APResolvers struct { + configRepository configrepository.ConfigRepository +} + +func New() *APResolvers { + return &APResolvers{ + configRepository: configrepository.Get(), + } +} + +var temporaryGlobalInstance *APResolvers + +func Get() *APResolvers { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} diff --git a/activitypub/webfinger/webfinger.go b/services/apfederation/webfinger/webfinger.go index 5cea75c87..f0482e332 100644 --- a/activitypub/webfinger/webfinger.go +++ b/services/apfederation/webfinger/webfinger.go @@ -8,8 +8,23 @@ import ( "strings" ) +type Webfinger struct{} + +var temporaryGlobalInstance *Webfinger + +func Get() *Webfinger { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} + +func New() *Webfinger { + return &Webfinger{} +} + // GetWebfingerLinks will return webfinger data for an account. -func GetWebfingerLinks(account string) ([]map[string]interface{}, error) { +func (w *Webfinger) GetWebfingerLinks(account string) ([]map[string]interface{}, error) { type webfingerResponse struct { Links []map[string]interface{} `json:"links"` } diff --git a/activitypub/workerpool/outbound.go b/services/apfederation/workerpool/outbound.go index a3d0fadea..402e91040 100644 --- a/activitypub/workerpool/outbound.go +++ b/services/apfederation/workerpool/outbound.go @@ -15,12 +15,29 @@ type Job struct { request *http.Request } -var queue chan Job +type WorkerPool struct { + queue chan Job +} -// InitOutboundWorkerPool starts n go routines that await ActivityPub jobs. -func InitOutboundWorkerPool() { - queue = make(chan Job) +func New() *WorkerPool { + wp := &WorkerPool{ + queue: make(chan Job), + } + wp.initOutboundWorkerPool() + return wp +} + +var temporaryGlobalInstance *WorkerPool +func Get() *WorkerPool { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + return temporaryGlobalInstance +} + +// InitOutboundWorkerPool starts n go routines that await ActivityPub jobs. +func (wp *WorkerPool) initOutboundWorkerPool() { // start workers for i := 1; i <= workerPoolSize; i++ { go worker(i, queue) @@ -28,23 +45,23 @@ func InitOutboundWorkerPool() { } // AddToOutboundQueue will queue up an outbound http request. -func AddToOutboundQueue(req *http.Request) { +func (wp *WorkerPool) AddToOutboundQueue(req *http.Request) { log.Tracef("Queued request for ActivityPub destination %s", req.RequestURI) - queue <- Job{req} + wp.queue <- Job{req} } -func worker(workerID int, queue <-chan Job) { +func (wp *WorkerPool) worker(workerID int, queue <-chan Job) { log.Debugf("Started ActivityPub worker %d", workerID) for job := range queue { - if err := sendActivityPubMessageToInbox(job); err != nil { + if err := wp.sendActivityPubMessageToInbox(job); err != nil { log.Errorf("ActivityPub destination %s failed to send Error: %s", job.request.RequestURI, err) } log.Tracef("Done with ActivityPub destination %s using worker %d", job.request.RequestURI, workerID) } } -func sendActivityPubMessageToInbox(job Job) error { +func (wp *WorkerPool) sendActivityPubMessageToInbox(job Job) error { client := &http.Client{} resp, err := client.Do(job.request) diff --git a/services/chat/chat.go b/services/chat/chat.go new file mode 100644 index 000000000..d3356d14e --- /dev/null +++ b/services/chat/chat.go @@ -0,0 +1,240 @@ +package chat + +import ( + "errors" + "fmt" + "net/http" + "sort" + + "github.com/owncast/owncast/models" + "github.com/owncast/owncast/storage/chatrepository" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/userrepository" + log "github.com/sirupsen/logrus" +) + +type Chat struct { + getStatus func() *models.Status + server *Server + configRepository *configrepository.SqlConfigRepository +} + +func New() *Chat { + return &Chat{ + configRepository: configrepository.Get(), + } +} + +var temporaryGlobalInstance *Chat + +// GetConfig returns the temporary global instance. +// Remove this after dependency injection is implemented. +func Get() *Chat { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + + return temporaryGlobalInstance +} + +// Start begins the chat server. +func (c *Chat) Start(getStatusFunc func() *models.Status) error { + c.getStatus = getStatusFunc + c.server = NewChat() + + go c.server.Run() + + log.Traceln("Chat server started with max connection count of", c.server.maxSocketConnectionLimit) + + return nil +} + +// GetClientsForUser will return chat connections that are owned by a specific user. +func (c *Chat) GetClientsForUser(userID string) ([]*Client, error) { + c.server.mu.Lock() + defer c.server.mu.Unlock() + + clients := map[string][]*Client{} + + for _, client := range c.server.clients { + clients[client.User.ID] = append(clients[client.User.ID], client) + } + + if _, exists := clients[userID]; !exists { + return nil, errors.New("no connections for user found") + } + + return clients[userID], nil +} + +// FindClientByID will return a single connected client by ID. +func (c *Chat) FindClientByID(clientID uint) (*Client, bool) { + client, found := c.server.clients[clientID] + return client, found +} + +// GetClients will return all the current chat clients connected. +func (c *Chat) GetClients() []*Client { + clients := []*Client{} + + if c.server == nil { + return clients + } + + // Convert the keyed map to a slice. + for _, client := range c.server.clients { + clients = append(clients, client) + } + + sort.Slice(clients, func(i, j int) bool { + return clients[i].ConnectedAt.Before(clients[j].ConnectedAt) + }) + + return clients +} + +// SendSystemMessage will send a message string as a system message to all clients. +func (c *Chat) SendSystemMessage(text string, ephemeral bool) error { + message := models.SystemMessageEvent{ + MessageEvent: models.MessageEvent{ + Body: text, + }, + } + message.SetDefaults() + message.RenderBody() + message.DisplayName = c.configRepository.GetServerName() + + if err := c.Broadcast(&message); err != nil { + log.Errorln("error sending system message", err) + } + + if !ephemeral { + cr := chatrepository.Get() + cr.SaveEvent(message.ID, nil, message.Body, message.GetMessageType(), nil, message.Timestamp, nil, nil, nil, nil) + } + + return nil +} + +// SendFediverseAction will send a message indicating some Fediverse engagement took place. +func (c *Chat) SendFediverseAction(eventType string, userAccountName string, image *string, body string, link string) error { + message := models.FediverseEngagementEvent{ + Event: models.Event{ + Type: eventType, + }, + MessageEvent: models.MessageEvent{ + Body: body, + }, + UserAccountName: userAccountName, + Image: image, + Link: link, + } + + message.SetDefaults() + message.RenderBody() + + if err := c.Broadcast(&message); err != nil { + log.Errorln("error sending system message", err) + return err + } + + cr := chatrepository.Get() + cr.SaveFederatedAction(message) + + return nil +} + +// SendSystemAction will send a system action string as an action event to all clients. +func (c *Chat) SendSystemAction(text string, ephemeral bool) error { + message := models.ActionEvent{ + MessageEvent: models.MessageEvent{ + Body: text, + }, + } + + message.SetDefaults() + message.RenderBody() + + if err := c.Broadcast(&message); err != nil { + log.Errorln("error sending system chat action") + } + + if !ephemeral { + cr := chatrepository.Get() + cr.SaveEvent(message.ID, nil, message.Body, message.GetMessageType(), nil, message.Timestamp, nil, nil, nil, nil) + } + + return nil +} + +// SendAllWelcomeMessage will send the chat message to all connected clients. +func (c *Chat) SendAllWelcomeMessage() { + c.server.sendAllWelcomeMessage() +} + +// SendSystemMessageToClient will send a single message to a single connected chat client. +func (c *Chat) SendSystemMessageToClient(clientID uint, text string) { + if client, foundClient := c.FindClientByID(clientID); foundClient { + c.server.sendSystemMessageToClient(client, text) + } +} + +// Broadcast will send all connected clients the outbound object provided. +func (c *Chat) Broadcast(event models.OutboundEvent) error { + return c.server.Broadcast(event.GetBroadcastPayload()) +} + +// HandleClientConnection handles a single inbound websocket connection. +func (c *Chat) HandleClientConnection(w http.ResponseWriter, r *http.Request) { + c.server.HandleClientConnection(w, r) +} + +// DisconnectClients will forcefully disconnect all clients belonging to a user by ID. +func (c *Chat) DisconnectClients(clients []*Client) { + c.server.DisconnectClients(clients) +} + +// SendConnectedClientInfoToUser will find all the connected clients assigned to a user +// and re-send each the connected client info. +func (c *Chat) SendConnectedClientInfoToUser(userID string) error { + clients, err := c.GetClientsForUser(userID) + if err != nil { + return err + } + + userRepository := userrepository.Get() + + // Get an updated reference to the user. + user := userRepository.GetUserByID(userID) + if user == nil { + return fmt.Errorf("user not found") + } + + if err != nil { + return err + } + + for _, client := range clients { + // Update the client's reference to its user. + client.User = user + // Send the update to the client. + client.sendConnectedClientInfo() + } + + return nil +} + +// SendActionToUser will send system action text to all connected clients +// assigned to a user ID. +func (c *Chat) SendActionToUser(userID string, text string) error { + clients, err := c.GetClientsForUser(userID) + if err != nil { + return err + } + + for _, client := range clients { + c.server.sendActionToClient(client, text) + } + + return nil +} diff --git a/core/chat/chatclient.go b/services/chat/chatclient.go index 8825d75cb..19dd6862c 100644 --- a/core/chat/chatclient.go +++ b/services/chat/chatclient.go @@ -11,7 +11,6 @@ import ( "golang.org/x/time/rate" "github.com/gorilla/websocket" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" "github.com/owncast/owncast/services/geoip" @@ -75,9 +74,9 @@ var ( ) func (c *Client) sendConnectedClientInfo() { - payload := events.ConnectedClientInfo{ - Event: events.Event{ - Type: events.ConnectedUserInfo, + payload := models.ConnectedClientInfo{ + Event: models.Event{ + Type: models.ConnectedUserInfo, }, User: c.User, } @@ -237,8 +236,8 @@ func (c *Client) sendPayload(payload interface{}) { } func (c *Client) sendAction(message string) { - clientMessage := events.ActionEvent{ - MessageEvent: events.MessageEvent{ + clientMessage := models.ActionEvent{ + MessageEvent: models.MessageEvent{ Body: message, }, } diff --git a/core/chat/events.go b/services/chat/events.go index 7f1678659..a61679523 100644 --- a/core/chat/events.go +++ b/services/chat/events.go @@ -6,16 +6,18 @@ import ( "strings" "time" - "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/services/webhooks" - "github.com/owncast/owncast/storage" + "github.com/owncast/owncast/storage/chatrepository" + "github.com/owncast/owncast/storage/userrepository" "github.com/owncast/owncast/utils" log "github.com/sirupsen/logrus" ) func (s *Server) userNameChanged(eventData chatClientEvent) { - var receivedEvent events.NameChangeEvent + var receivedEvent models.NameChangeEvent if err := json.Unmarshal(eventData.data, &receivedEvent); err != nil { log.Errorln("error unmarshalling to NameChangeEvent", err) return @@ -24,8 +26,8 @@ func (s *Server) userNameChanged(eventData chatClientEvent) { proposedUsername := receivedEvent.NewName // Check if name is on the blocklist - blocklist := configRepository.GetForbiddenUsernameList() - userRepository := storage.GetUserRepository() + blocklist := s.configRepository.GetForbiddenUsernameList() + userRepository := userrepository.Get() // Names have a max length proposedUsername = utils.MakeSafeStringOfLength(proposedUsername, config.MaxChatDisplayNameLength) @@ -82,7 +84,7 @@ func (s *Server) userNameChanged(eventData chatClientEvent) { // Send chat event letting everyone about about the name change savedUser.DisplayName = proposedUsername - broadcastEvent := events.NameChangeBroadcast{ + broadcastEvent := models.NameChangeBroadcast{ Oldname: oldName, } broadcastEvent.User = savedUser @@ -104,7 +106,7 @@ func (s *Server) userNameChanged(eventData chatClientEvent) { } func (s *Server) userColorChanged(eventData chatClientEvent) { - var receivedEvent events.ColorChangeEvent + var receivedEvent models.ColorChangeEvent if err := json.Unmarshal(eventData.data, &receivedEvent); err != nil { log.Errorln("error unmarshalling to ColorChangeEvent", err) return @@ -115,7 +117,7 @@ func (s *Server) userColorChanged(eventData chatClientEvent) { log.Errorln("invalid color requested when changing user display color") return } - userRepository := storage.GetUserRepository() + userRepository := userrepository.Get() // Save the new color if err := userRepository.ChangeUserColor(eventData.client.User.ID, receivedEvent.NewColor); err != nil { @@ -128,7 +130,7 @@ func (s *Server) userColorChanged(eventData chatClientEvent) { } func (s *Server) userMessageSent(eventData chatClientEvent) { - var event events.UserMessageEvent + var event models.UserMessageEvent if err := json.Unmarshal(eventData.data, &event); err != nil { log.Errorln("error unmarshalling to UserMessageEvent", err) return @@ -142,15 +144,17 @@ func (s *Server) userMessageSent(eventData chatClientEvent) { return } + st := status.Get() + // Ignore if the stream has been offline - if !getStatus().Online && getStatus().LastDisconnectTime != nil { - disconnectedTime := getStatus().LastDisconnectTime.Time + if st.Online && st.Status.LastDisconnectTime != nil { + disconnectedTime := st.Status.LastDisconnectTime.Time if time.Since(disconnectedTime) > 5*time.Minute { return } } - userRepository := storage.GetUserRepository() + userRepository := userrepository.Get() event.User = userRepository.GetUserByToken(eventData.client.accessToken) @@ -168,11 +172,12 @@ func (s *Server) userMessageSent(eventData chatClientEvent) { // Send chat message sent webhook webhookManager := webhooks.Get() webhookManager.SendChatEvent(&event) - chatMessagesSentCounter.Inc() + s.chatMessagesSentCounter.Inc() - SaveUserMessage(event) + cr := chatrepository.Get() + cr.SaveUserMessage(event) eventData.client.MessageCount++ - _lastSeenCache[event.User.ID] = time.Now() + s.lastSeenCache[event.User.ID] = time.Now() } func logSanitize(userValue string) string { diff --git a/core/chat/messageRendering_test.go b/services/chat/messageRendering_test.go index 382d83cdb..5a8ea0a96 100644 --- a/core/chat/messageRendering_test.go +++ b/services/chat/messageRendering_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" ) // Test a bunch of arbitrary markup and markdown to make sure we get sanitized @@ -54,7 +55,7 @@ func TestAllowEmojiImages(t *testing.T) { func TestAllowHTML(t *testing.T) { messageContent := `<img src="/img/emoji/beerparrot.gif"><ul><li>**test thing**</li></ul>` expected := "<p><img src=\"/img/emoji/beerparrot.gif\"><ul><li><strong>test thing</strong></li></ul></p>\n" - result := events.RenderMarkdown(messageContent) + result := models.RenderMarkdown(messageContent) if result != expected { t.Errorf("message rendering does not match expected. Got\n%s, \n\n want:\n%s", result, expected) diff --git a/core/chat/messages.go b/services/chat/messages.go index 91316b4b3..ddb63f19d 100644 --- a/core/chat/messages.go +++ b/services/chat/messages.go @@ -3,29 +3,32 @@ package chat import ( "errors" - "github.com/owncast/owncast/core/chat/events" + "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/webhooks" + "github.com/owncast/owncast/storage/chatrepository" log "github.com/sirupsen/logrus" ) // SetMessagesVisibility will set the visibility of multiple messages by ID. -func SetMessagesVisibility(messageIDs []string, visibility bool) error { +func (c *Chat) SetMessagesVisibility(messageIDs []string, visibility bool) error { + cr := chatrepository.Get() + // Save new message visibility - if err := saveMessageVisibility(messageIDs, visibility); err != nil { + if err := cr.SaveMessageVisibility(messageIDs, visibility); err != nil { log.Errorln(err) return err } // Send an event letting the chat clients know to hide or show // the messages. - event := events.SetMessageVisibilityEvent{ + event := models.SetMessageVisibilityEvent{ MessageIDs: messageIDs, Visible: visibility, } event.Event.SetDefaults() payload := event.GetBroadcastPayload() - if err := _server.Broadcast(payload); err != nil { + if err := c.server.Broadcast(payload); err != nil { return errors.New("error broadcasting message visibility payload " + err.Error()) } diff --git a/core/chat/server.go b/services/chat/server.go index 8d81fcd6c..0396a996e 100644 --- a/core/chat/server.go +++ b/services/chat/server.go @@ -7,24 +7,23 @@ import ( "sync" "time" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" log "github.com/sirupsen/logrus" "github.com/gorilla/websocket" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" "github.com/owncast/owncast/services/geoip" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/services/webhooks" - "github.com/owncast/owncast/storage" + "github.com/owncast/owncast/storage/chatrepository" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/userrepository" "github.com/owncast/owncast/utils" ) -var _server *Server - -// a map of user IDs and when they last were active. -var _lastSeenCache = map[string]time.Time{} - // Server represents an instance of the chat server. type Server struct { clients map[uint]*Client @@ -41,24 +40,41 @@ type Server struct { geoipClient *geoip.Client seq uint maxSocketConnectionLimit int64 + chatMessagesSentCounter prometheus.Gauge + + // a map of user IDs and when they last were active. + lastSeenCache map[string]time.Time - mu sync.RWMutex + mu sync.RWMutex + config *config.Config + configRepository *configrepository.SqlConfigRepository + chatRepository *chatrepository.ChatRepository } // NewChat will return a new instance of the chat server. func NewChat() *Server { - maximumConcurrentConnectionLimit := getMaximumConcurrentConnectionLimit() - setSystemConcurrentConnectionLimit(maximumConcurrentConnectionLimit) - server := &Server{ clients: map[uint]*Client{}, outbound: make(chan []byte), inbound: make(chan chatClientEvent), unregister: make(chan uint), - maxSocketConnectionLimit: maximumConcurrentConnectionLimit, + maxSocketConnectionLimit: 100, // TODO: Set this properly! + lastSeenCache: map[string]time.Time{}, geoipClient: geoip.NewClient(), + config: config.Get(), + configRepository: configrepository.Get(), + chatRepository: chatrepository.Get(), } + server.chatMessagesSentCounter = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "total_chat_message_count", + Help: "The number of chat messages incremented over time.", + ConstLabels: map[string]string{ + "version": server.config.VersionNumber, + "host": server.configRepository.GetServerURL(), + }, + }) + return server } @@ -93,8 +109,8 @@ func (s *Server) Addclient(conn *websocket.Conn, user *models.User, accessToken } // Do not send user re-joined broadcast message if they've been active within 10 minutes. - shouldSendJoinedMessages := configRepository.GetChatJoinMessagesEnabled() - if previouslyLastSeen, ok := _lastSeenCache[user.ID]; ok && time.Since(previouslyLastSeen) < time.Minute*10 { + shouldSendJoinedMessages := s.configRepository.GetChatJoinMessagesEnabled() + if previouslyLastSeen, ok := s.lastSeenCache[user.ID]; ok && time.Since(previouslyLastSeen) < time.Minute*10 { shouldSendJoinedMessages = false } @@ -103,7 +119,7 @@ func (s *Server) Addclient(conn *websocket.Conn, user *models.User, accessToken client.Id = s.seq s.clients[client.Id] = client s.seq++ - _lastSeenCache[user.ID] = time.Now() + s.lastSeenCache[user.ID] = time.Now() } s.mu.Unlock() @@ -114,7 +130,9 @@ func (s *Server) Addclient(conn *websocket.Conn, user *models.User, accessToken client.sendConnectedClientInfo() - if getStatus().Online { + st := status.Get() + + if st.Online { if shouldSendJoinedMessages { s.sendUserJoinedMessage(client) } @@ -130,7 +148,7 @@ func (s *Server) Addclient(conn *websocket.Conn, user *models.User, accessToken } func (s *Server) sendUserJoinedMessage(c *Client) { - userJoinedEvent := events.UserJoinedEvent{} + userJoinedEvent := models.UserJoinedEvent{} userJoinedEvent.SetDefaults() userJoinedEvent.User = c.User userJoinedEvent.ClientID = c.Id @@ -158,14 +176,17 @@ func (s *Server) ClientClosed(c *Client) { // HandleClientConnection is fired when a single client connects to the websocket. func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) { - if configRepository.GetChatDisabled() { - _, _ = w.Write([]byte(events.ChatDisabled)) + cr := configrepository.Get() + chatRepository := chatrepository.Get() + + if cr.GetChatDisabled() { + _, _ = w.Write([]byte(models.ChatDisabled)) return } ipAddress := utils.GetIPAddressFromRequest(r) // Check if this client's IP address is banned. If so send a rejection. - if blocked, err := configRepository.IsIPAddressBanned(ipAddress); blocked { + if blocked, err := chatRepository.IsIPAddressBanned(ipAddress); blocked { log.Debugln("Client ip address has been blocked. Rejecting.") w.WriteHeader(http.StatusForbidden) @@ -177,7 +198,7 @@ func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) // Limit concurrent chat connections if int64(len(s.clients)) >= s.maxSocketConnectionLimit { log.Warnln("rejecting incoming client connection as it exceeds the max client count of", s.maxSocketConnectionLimit) - _, _ = w.Write([]byte(events.ErrorMaxConnectionsExceeded)) + _, _ = w.Write([]byte(models.ErrorMaxConnectionsExceeded)) return } @@ -200,14 +221,14 @@ func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) return } - userRepository := storage.GetUserRepository() + userRepository := userrepository.Get() // A user is required to use the websocket user := userRepository.GetUserByToken(accessToken) if user == nil { // Send error that registration is required - _ = conn.WriteJSON(events.EventPayload{ - "type": events.ErrorNeedsRegistration, + _ = conn.WriteJSON(models.EventPayload{ + "type": models.ErrorNeedsRegistration, }) _ = conn.Close() return @@ -216,8 +237,8 @@ func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) // User is disabled therefore we should disconnect. if user.DisabledAt != nil { log.Traceln("Disabled user", user.ID, user.DisplayName, "rejected") - _ = conn.WriteJSON(events.EventPayload{ - "type": events.ErrorUserDisabled, + _ = conn.WriteJSON(models.EventPayload{ + "type": models.ErrorUserDisabled, }) _ = conn.Close() return @@ -229,7 +250,7 @@ func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) } // Broadcast sends message to all connected clients. -func (s *Server) Broadcast(payload events.EventPayload) error { +func (s *Server) Broadcast(payload models.EventPayload) error { data, err := json.Marshal(payload) if err != nil { return err @@ -254,7 +275,7 @@ func (s *Server) Broadcast(payload events.EventPayload) error { } // Send will send a single payload to a single connected client. -func (s *Server) Send(payload events.EventPayload, client *Client) { +func (s *Server) Send(payload models.EventPayload, client *Client) { data, err := json.Marshal(payload) if err != nil { log.Errorln(err) @@ -270,12 +291,12 @@ func (s *Server) DisconnectClients(clients []*Client) { log.Traceln("Disconnecting client", client.User.ID, "owned by", client.User.DisplayName) go func(client *Client) { - event := events.UserDisabledEvent{} + event := models.UserDisabledEvent{} event.SetDefaults() // Send this disabled event specifically to this single connected client // to let them know they've been banned. - _server.Send(event.GetBroadcastPayload(), client) + s.Send(event.GetBroadcastPayload(), client) // Give the socket time to send out the above message. // Unfortunately I don't know of any way to get a real callback to know when @@ -290,58 +311,15 @@ func (s *Server) DisconnectClients(clients []*Client) { } } -// SendConnectedClientInfoToUser will find all the connected clients assigned to a user -// and re-send each the connected client info. -func SendConnectedClientInfoToUser(userID string) error { - clients, err := GetClientsForUser(userID) - if err != nil { - return err - } - - userRepository := storage.GetUserRepository() - - // Get an updated reference to the user. - user := userRepository.GetUserByID(userID) - if user == nil { - return fmt.Errorf("user not found") - } - - if err != nil { - return err - } - - for _, client := range clients { - // Update the client's reference to its user. - client.User = user - // Send the update to the client. - client.sendConnectedClientInfo() - } - - return nil -} - -// SendActionToUser will send system action text to all connected clients -// assigned to a user ID. -func SendActionToUser(userID string, text string) error { - clients, err := GetClientsForUser(userID) - if err != nil { - return err - } - - for _, client := range clients { - _server.sendActionToClient(client, text) - } - - return nil -} - func (s *Server) eventReceived(event chatClientEvent) { c := event.client u := c.User + cr := configrepository.Get() + // If established chat user only mode is enabled and the user is not old // enough then reject this event and send them an informative message. - if u != nil && configRepository.GetChatEstbalishedUsersOnlyMode() && time.Since(event.client.User.CreatedAt) < config.GetDefaults().ChatEstablishedUserModeTimeDuration && !u.IsModerator() { + if u != nil && cr.GetChatEstbalishedUsersOnlyMode() && time.Since(event.client.User.CreatedAt) < config.GetDefaults().ChatEstablishedUserModeTimeDuration && !u.IsModerator() { s.sendActionToClient(c, "You have not been an established chat participant long enough to take part in chat. Please enjoy the stream and try again later.") return } @@ -354,13 +332,13 @@ func (s *Server) eventReceived(event chatClientEvent) { eventType := typecheck["type"] switch eventType { - case events.MessageSent: + case models.MessageSent: s.userMessageSent(event) - case events.UserNameChanged: + case models.UserNameChanged: s.userNameChanged(event) - case events.UserColorChanged: + case models.UserColorChanged: s.userColorChanged(event) default: log.Debugln(logSanitize(fmt.Sprint(eventType)), "event not found:", logSanitize(fmt.Sprint(typecheck))) @@ -370,8 +348,9 @@ func (s *Server) eventReceived(event chatClientEvent) { func (s *Server) sendWelcomeMessageToClient(c *Client) { // Add an artificial delay so people notice this message come in. time.Sleep(7 * time.Second) + cr := configrepository.Get() - welcomeMessage := utils.RenderSimpleMarkdown(configRepository.GetServerWelcomeMessage()) + welcomeMessage := utils.RenderSimpleMarkdown(cr.GetServerWelcomeMessage()) if welcomeMessage != "" { s.sendSystemMessageToClient(c, welcomeMessage) @@ -379,39 +358,42 @@ func (s *Server) sendWelcomeMessageToClient(c *Client) { } func (s *Server) sendAllWelcomeMessage() { - welcomeMessage := utils.RenderSimpleMarkdown(configRepository.GetServerWelcomeMessage()) + cr := configrepository.Get() + welcomeMessage := utils.RenderSimpleMarkdown(cr.GetServerWelcomeMessage()) if welcomeMessage != "" { - clientMessage := events.SystemMessageEvent{ - Event: events.Event{}, - MessageEvent: events.MessageEvent{ + clientMessage := models.SystemMessageEvent{ + Event: models.Event{}, + MessageEvent: models.MessageEvent{ Body: welcomeMessage, }, } clientMessage.SetDefaults() + clientMessage.DisplayName = s.configRepository.GetServerName() _ = s.Broadcast(clientMessage.GetBroadcastPayload()) } } func (s *Server) sendSystemMessageToClient(c *Client, message string) { - clientMessage := events.SystemMessageEvent{ - Event: events.Event{}, - MessageEvent: events.MessageEvent{ + clientMessage := models.SystemMessageEvent{ + Event: models.Event{}, + MessageEvent: models.MessageEvent{ Body: message, }, } clientMessage.SetDefaults() clientMessage.RenderBody() + clientMessage.DisplayName = s.configRepository.GetServerName() s.Send(clientMessage.GetBroadcastPayload(), c) } func (s *Server) sendActionToClient(c *Client, message string) { - clientMessage := events.ActionEvent{ - MessageEvent: events.MessageEvent{ + clientMessage := models.ActionEvent{ + MessageEvent: models.MessageEvent{ Body: message, }, - Event: events.Event{ - Type: events.ChatActionSent, + Event: models.Event{ + Type: models.ChatActionSent, }, } clientMessage.SetDefaults() diff --git a/services/config/config.go b/services/config/config.go index c58a17483..0188dcd7c 100644 --- a/services/config/config.go +++ b/services/config/config.go @@ -44,7 +44,7 @@ type Config struct { } // NewFediAuth creates a new FediAuth instance. -func NewConfig() *Config { +func New() *Config { // Default config values. c := &Config{ DatabaseFilePath: "data/owncast.db", @@ -71,9 +71,9 @@ var temporaryGlobalInstance *Config // GetConfig returns the temporary global instance. // Remove this after dependency injection is implemented. -func GetConfig() *Config { +func Get() *Config { if temporaryGlobalInstance == nil { - temporaryGlobalInstance = NewConfig() + temporaryGlobalInstance = New() } return temporaryGlobalInstance @@ -104,3 +104,8 @@ func (c *Config) GetReleaseString() string { return fmt.Sprintf("Owncast v%s-%s (%s)", versionNumber, buildPlatform, gitCommit) } + +// GetTranscoderLogFilePath returns the logging path for the transcoder log output. +func (c *Config) GetTranscoderLogFilePath() string { + return filepath.Join(c.LogDirectory, "transcoder.log") +} diff --git a/services/metrics/healthOverview.go b/services/metrics/healthOverview.go index 0318874dd..64ddf5827 100644 --- a/services/metrics/healthOverview.go +++ b/services/metrics/healthOverview.go @@ -6,6 +6,7 @@ import ( "github.com/owncast/owncast/core" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/utils" ) @@ -16,8 +17,6 @@ const ( minClientCountForDetails = 3 ) -var configRepository = configrepository.Get() - // GetStreamHealthOverview will return the stream health overview. func (m *Metrics) GetStreamHealthOverview() *models.StreamHealthOverview { return m.metrics.streamHealthOverview @@ -71,6 +70,8 @@ func (m *Metrics) networkSpeedHealthOverviewMessage() string { bitrate int } + configRepository := configrepository.Get() + outputVariants := configRepository.GetStreamOutputVariants() streamSortVariants := make([]singleVariant, len(outputVariants)) @@ -137,12 +138,14 @@ func (m *Metrics) wastefulBitrateOverviewMessage() string { return "" } - currentBroadcast := core.GetCurrentBroadcast() + stat := status.Get() + + currentBroadcast := stat.GetCurrentBroadcast() if currentBroadcast == nil { return "" } - currentBroadcaster := core.GetBroadcaster() + currentBroadcaster := stat.GetBroadcaster() if currentBroadcast == nil { return "" } @@ -156,6 +159,7 @@ func (m *Metrics) wastefulBitrateOverviewMessage() string { if inboundBitrate == 0 { return "" } + configRepository := configrepository.Get() outputVariants := configRepository.GetStreamOutputVariants() @@ -230,6 +234,8 @@ func (m *Metrics) errorCountHealthOverviewMessage() string { if totalNumberOfClients >= minClientCountForDetails { healthyPercentage := utils.IntPercentage(clientsWithErrors, totalNumberOfClients) + configRepository := configrepository.Get() + isUsingPassthrough := false outputVariants := configRepository.GetStreamOutputVariants() for _, variant := range outputVariants { @@ -242,7 +248,8 @@ func (m *Metrics) errorCountHealthOverviewMessage() string { return fmt.Sprintf("%d of %d viewers (%d%%) are experiencing errors. You're currently using a video passthrough output, often known for causing playback issues for people. It is suggested you turn it off.", clientsWithErrors, totalNumberOfClients, healthyPercentage) } - currentBroadcast := core.GetCurrentBroadcast() + stat := status.Get() + currentBroadcast := stat.GetCurrentBroadcast() if currentBroadcast != nil && currentBroadcast.LatencyLevel.SecondsPerSegment < 3 { return fmt.Sprintf("%d of %d viewers (%d%%) may be experiencing some issues. You may want to increase your latency buffer level in your video configuration to see if it helps.", clientsWithErrors, totalNumberOfClients, healthyPercentage) } diff --git a/services/metrics/metrics.go b/services/metrics/metrics.go index f17c0b092..a6417576c 100644 --- a/services/metrics/metrics.go +++ b/services/metrics/metrics.go @@ -5,7 +5,9 @@ import ( "time" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/chat" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" "github.com/prometheus/client_golang/prometheus" ) @@ -26,6 +28,8 @@ type Metrics struct { chatUserCount prometheus.Gauge currentChatMessageCount prometheus.Gauge playbackErrorCount prometheus.Gauge + + chatService *chat.Chat } // How often we poll for updates. @@ -76,6 +80,7 @@ func New() *Metrics { windowedBandwidths: map[string]float64{}, windowedLatencies: map[string]float64{}, windowedDownloadDurations: map[string]float64{}, + chatService: chat.Get(), } } @@ -84,12 +89,14 @@ func New() *Metrics { // Start will begin the metrics collection and alerting. func (m *Metrics) Start(getStatus func() models.Status) { m.getStatus = getStatus + configRepository := configrepository.Get() + host := configRepository.GetServerURL() if host == "" { host = "unknown" } - c := config.GetConfig() + c := config.Get() m.labels = map[string]string{ "version": c.VersionNumber, diff --git a/services/metrics/playback.go b/services/metrics/playback.go index 7263a463f..31ee36600 100644 --- a/services/metrics/playback.go +++ b/services/metrics/playback.go @@ -6,6 +6,7 @@ import ( "github.com/owncast/owncast/core" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/utils" ) @@ -13,8 +14,10 @@ func (m *Metrics) handlePlaybackPolling() { m.metrics.m.Lock() defer m.metrics.m.Unlock() + s := status.Get() + // Make sure this is fired first before all the values get cleared below. - if m.getStatus().Online { + if s.Online { m.generateStreamHealthOverview() } diff --git a/services/metrics/viewers.go b/services/metrics/viewers.go index 912d4c1b1..5e243cbc9 100644 --- a/services/metrics/viewers.go +++ b/services/metrics/viewers.go @@ -4,9 +4,8 @@ import ( "time" "github.com/nakabonne/tstorage" - "github.com/owncast/owncast/core" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/storage/chatrepository" "github.com/owncast/owncast/storage/userrepository" log "github.com/sirupsen/logrus" @@ -30,12 +29,14 @@ func (m *Metrics) startViewerCollectionMetrics() { } func (m *Metrics) collectViewerCount() { + s := status.Get() + // Don't collect metrics for viewers if there's no stream active. - if !core.GetStatus().Online { + if !s.Online { return } - count := core.GetStatus().ViewerCount + count := s.ViewerCount // Save active viewer count to our Prometheus collector. m.activeViewerCount.Set(float64(count)) @@ -52,9 +53,9 @@ func (m *Metrics) collectViewerCount() { } func (m *Metrics) collectChatClientCount() { - count := len(chat.GetClients()) + count := len(m.chatService.GetClients()) m.activeChatClientCount.Set(float64(count)) - chatRepository := chatrepository.GetChatRepository() + chatRepository := chatrepository.Get() usersRepository := userrepository.Get() // Total message count diff --git a/services/notifications/notifications.go b/services/notifications/notifications.go index efb650770..f8dd313bc 100644 --- a/services/notifications/notifications.go +++ b/services/notifications/notifications.go @@ -22,12 +22,15 @@ type Notifier struct { } var ( - configRepository = configrepository.Get() - notificationsRepository = notificationsrepository.Get() + configRepository *configrepository.SqlConfigRepository + notificationsRepository *notificationsrepository.SqlNotificationsRepository ) // Setup will perform any pre-use setup for the notifier. func Setup(datastore *data.Store) { + configRepository = configrepository.Get() + notificationsRepository = notificationsrepository.Get() + initializeBrowserPushIfNeeded() } diff --git a/services/status/status.go b/services/status/status.go new file mode 100644 index 000000000..611ebee4d --- /dev/null +++ b/services/status/status.go @@ -0,0 +1,61 @@ +package status + +import ( + "github.com/owncast/owncast/models" + "github.com/owncast/owncast/storage/configrepository" +) + +type Status struct { + models.Stats + models.Status + + broadcast *models.CurrentBroadcast + broadcaster *models.Broadcaster + StreamConnected bool + + VersionNumber string `json:"versionNumber"` + StreamTitle string `json:"streamTitle"` + ViewerCount int `json:"viewerCount"` + OverallMaxViewerCount int `json:"overallMaxViewerCount"` + SessionMaxViewerCount int `json:"sessionMaxViewerCount"` + + Online bool `json:"online"` +} + +var temporaryGlobalInstance *Status + +func New() *Status { + configRepository := configrepository.Get() + + return &Status{ + StreamTitle: configRepository.GetStreamTitle(), + } +} + +// Get will return the global instance of the status service. +func Get() *Status { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = &Status{} + } + + return temporaryGlobalInstance +} + +// GetCurrentBroadcast will return the currently active broadcast. +func (s *Status) GetCurrentBroadcast() *models.CurrentBroadcast { + return s.broadcast +} + +func (s *Status) SetCurrentBroadcast(broadcast *models.CurrentBroadcast) { + s.broadcast = broadcast +} + +// SetBroadcaster will store the current inbound broadcasting details. +func (s *Status) SetBroadcaster(broadcaster *models.Broadcaster) { + s.broadcaster = broadcaster +} + +// GetBroadcaster will return the details of the currently active broadcaster. +func (s *Status) GetBroadcaster() *models.Broadcaster { + return s.broadcaster +} diff --git a/services/webhooks/chat.go b/services/webhooks/chat.go index 418d5d34f..8b6be69b0 100644 --- a/services/webhooks/chat.go +++ b/services/webhooks/chat.go @@ -1,12 +1,11 @@ package webhooks import ( - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" ) // SendChatEvent will send a chat event to webhook destinations. -func (w *LiveWebhookManager) SendChatEvent(chatEvent *events.UserMessageEvent) { +func (w *LiveWebhookManager) SendChatEvent(chatEvent *models.UserMessageEvent) { webhookEvent := WebhookEvent{ Type: chatEvent.GetMessageType(), EventData: &WebhookChatMessage{ @@ -24,7 +23,7 @@ func (w *LiveWebhookManager) SendChatEvent(chatEvent *events.UserMessageEvent) { } // SendChatEventUsernameChanged will send a username changed event to webhook destinations. -func (w *LiveWebhookManager) SendChatEventUsernameChanged(event events.NameChangeEvent) { +func (w *LiveWebhookManager) SendChatEventUsernameChanged(event models.NameChangeEvent) { webhookEvent := WebhookEvent{ Type: models.UserNameChanged, EventData: event, @@ -34,7 +33,7 @@ func (w *LiveWebhookManager) SendChatEventUsernameChanged(event events.NameChang } // SendChatEventUserJoined sends a webhook notifying that a user has joined. -func (w *LiveWebhookManager) SendChatEventUserJoined(event events.UserJoinedEvent) { +func (w *LiveWebhookManager) SendChatEventUserJoined(event models.UserJoinedEvent) { webhookEvent := WebhookEvent{ Type: models.UserJoined, EventData: event, @@ -45,7 +44,7 @@ func (w *LiveWebhookManager) SendChatEventUserJoined(event events.UserJoinedEven // SendChatEventSetMessageVisibility sends a webhook notifying that the visibility of one or more // messages has changed. -func (w *LiveWebhookManager) SendChatEventSetMessageVisibility(event events.SetMessageVisibilityEvent) { +func (w *LiveWebhookManager) SendChatEventSetMessageVisibility(event models.SetMessageVisibilityEvent) { webhookEvent := WebhookEvent{ Type: models.VisibiltyToggled, EventData: event, diff --git a/services/webhooks/chat_test.go b/services/webhooks/chat_test.go index 9ebc09ca2..91a2e1ca6 100644 --- a/services/webhooks/chat_test.go +++ b/services/webhooks/chat_test.go @@ -4,14 +4,12 @@ import ( "testing" "time" - "github.com/owncast/owncast/core/chat/events" - "github.com/owncast/owncast/core/user" "github.com/owncast/owncast/models" ) func TestSendChatEvent(t *testing.T) { timestamp := time.Unix(72, 6).UTC() - user := user.User{ + user := models.User{ ID: "user id", DisplayName: "display name", DisplayColor: 4, @@ -26,18 +24,18 @@ func TestSendChatEvent(t *testing.T) { } checkPayload(t, models.MessageSent, func() { - manager.SendChatEvent(&events.UserMessageEvent{ - Event: events.Event{ - Type: events.MessageSent, + manager.SendChatEvent(&models.UserMessageEvent{ + Event: models.Event{ + Type: models.MessageSent, ID: "id", Timestamp: timestamp, }, - UserEvent: events.UserEvent{ + UserEvent: models.UserEvent{ User: &user, ClientID: 51, HiddenAt: nil, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ OutboundEvent: nil, Body: "body", RawBody: "raw body", @@ -64,7 +62,7 @@ func TestSendChatEvent(t *testing.T) { func TestSendChatEventUsernameChanged(t *testing.T) { timestamp := time.Unix(72, 6).UTC() - user := user.User{ + user := models.User{ ID: "user id", DisplayName: "display name", DisplayColor: 4, @@ -79,13 +77,13 @@ func TestSendChatEventUsernameChanged(t *testing.T) { } checkPayload(t, models.UserNameChanged, func() { - manager.SendChatEventUsernameChanged(events.NameChangeEvent{ - Event: events.Event{ - Type: events.UserNameChanged, + manager.SendChatEventUsernameChanged(models.NameChangeEvent{ + Event: models.Event{ + Type: models.UserNameChanged, ID: "id", Timestamp: timestamp, }, - UserEvent: events.UserEvent{ + UserEvent: models.UserEvent{ User: &user, ClientID: 51, HiddenAt: nil, @@ -112,7 +110,7 @@ func TestSendChatEventUsernameChanged(t *testing.T) { func TestSendChatEventUserJoined(t *testing.T) { timestamp := time.Unix(72, 6).UTC() - user := user.User{ + user := models.User{ ID: "user id", DisplayName: "display name", DisplayColor: 4, @@ -127,13 +125,13 @@ func TestSendChatEventUserJoined(t *testing.T) { } checkPayload(t, models.UserJoined, func() { - manager.SendChatEventUserJoined(events.UserJoinedEvent{ - Event: events.Event{ - Type: events.UserJoined, + manager.SendChatEventUserJoined(models.UserJoinedEvent{ + Event: models.Event{ + Type: models.UserJoined, ID: "id", Timestamp: timestamp, }, - UserEvent: events.UserEvent{ + UserEvent: models.UserEvent{ User: &user, ClientID: 51, HiddenAt: nil, @@ -160,13 +158,13 @@ func TestSendChatEventSetMessageVisibility(t *testing.T) { timestamp := time.Unix(72, 6).UTC() checkPayload(t, models.VisibiltyToggled, func() { - manager.SendChatEventSetMessageVisibility(events.SetMessageVisibilityEvent{ - Event: events.Event{ - Type: events.VisibiltyUpdate, + manager.SendChatEventSetMessageVisibility(models.SetMessageVisibilityEvent{ + Event: models.Event{ + Type: models.VisibiltyUpdate, ID: "id", Timestamp: timestamp, }, - UserMessageEvent: events.UserMessageEvent{}, + UserMessageEvent: models.UserMessageEvent{}, MessageIDs: []string{"message1", "message2"}, Visible: false, }) diff --git a/services/webhooks/manager.go b/services/webhooks/manager.go index 59a4de244..055c6fb78 100644 --- a/services/webhooks/manager.go +++ b/services/webhooks/manager.go @@ -10,11 +10,11 @@ type Manager interface { // to be sent out to all registered webhook destinations. type LiveWebhookManager struct { queue chan Job - getStatus func() models.Status + getStatus func() *models.Status } // New creates a new webhook manager. -func New(getStatusFunc func() models.Status) *LiveWebhookManager { +func New(getStatusFunc func() *models.Status) *LiveWebhookManager { m := &LiveWebhookManager{ getStatus: getStatusFunc, } @@ -24,7 +24,7 @@ func New(getStatusFunc func() models.Status) *LiveWebhookManager { // InitTemporarySingleton initializes the the temporary global instance of the webhook manager // to be deleted once dependency injection is implemented. -func InitTemporarySingleton(getStatusFunc func() models.Status) { +func InitTemporarySingleton(getStatusFunc func() *models.Status) { temporaryGlobalInstance = New(getStatusFunc) } diff --git a/services/webhooks/stream.go b/services/webhooks/stream.go index 75ed82b4d..bc111ff84 100644 --- a/services/webhooks/stream.go +++ b/services/webhooks/stream.go @@ -8,14 +8,14 @@ import ( "github.com/teris-io/shortid" ) -var configRepository = configrepository.Get() - // SendStreamStatusEvent will send all webhook destinations the current stream status. func (w *LiveWebhookManager) SendStreamStatusEvent(eventType models.EventType) { w.sendStreamStatusEvent(eventType, shortid.MustGenerate(), time.Now()) } func (w *LiveWebhookManager) sendStreamStatusEvent(eventType models.EventType, id string, timestamp time.Time) { + configRepository := configrepository.Get() + w.SendEventToWebhooks(WebhookEvent{ Type: eventType, EventData: map[string]interface{}{ diff --git a/services/webhooks/stream_test.go b/services/webhooks/stream_test.go index b4df5fcc2..0dbca3482 100644 --- a/services/webhooks/stream_test.go +++ b/services/webhooks/stream_test.go @@ -4,17 +4,19 @@ import ( "testing" "time" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/storage/configrepository" ) func TestSendStreamStatusEvent(t *testing.T) { + configRepository := configrepository.Get() + configRepository.SetServerName("my server") configRepository.SetServerSummary("my server where I stream") configRepository.SetStreamTitle("my stream") checkPayload(t, models.StreamStarted, func() { - manager.sendStreamStatusEvent(events.StreamStarted, "id", time.Unix(72, 6).UTC()) + manager.sendStreamStatusEvent(models.StreamStarted, "id", time.Unix(72, 6).UTC()) }, `{ "id": "id", "name": "my server", diff --git a/services/webhooks/webhooks.go b/services/webhooks/webhooks.go index 2cd7dcf5f..439344915 100644 --- a/services/webhooks/webhooks.go +++ b/services/webhooks/webhooks.go @@ -25,14 +25,13 @@ type WebhookChatMessage struct { Visible bool `json:"visible"` } -var webhookRepository = storage.GetWebhookRepository() - // SendEventToWebhooks will send a single webhook event to all webhook destinations. func (w *LiveWebhookManager) SendEventToWebhooks(payload WebhookEvent) { w.sendEventToWebhooks(payload, nil) } func (w *LiveWebhookManager) sendEventToWebhooks(payload WebhookEvent, wg *sync.WaitGroup) { + webhookRepository := storage.GetWebhookRepository() webhooks := webhookRepository.GetWebhooksForEvent(payload.Type) for _, webhook := range webhooks { diff --git a/services/webhooks/webhooks_test.go b/services/webhooks/webhooks_test.go index 5094f4b11..73830fe8b 100644 --- a/services/webhooks/webhooks_test.go +++ b/services/webhooks/webhooks_test.go @@ -6,20 +6,21 @@ import ( "fmt" "net/http" "net/http/httptest" - "os" "sync" "sync/atomic" "testing" "time" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/storage" + "github.com/owncast/owncast/storage/data" jsonpatch "gopkg.in/evanphx/json-patch.v5" ) var manager *LiveWebhookManager -func fakeGetStatus() models.Status { - return models.Status{ +func fakeGetStatus() *models.Status { + return &models.Status{ Online: true, ViewerCount: 5, OverallMaxViewerCount: 420, @@ -30,16 +31,10 @@ func fakeGetStatus() models.Status { } func TestMain(m *testing.M) { - dbFile, err := os.CreateTemp(os.TempDir(), "owncast-test-db.db") + _, err := data.NewStore(":memory:") if err != nil { panic(err) } - dbFile.Close() - defer os.Remove(dbFile.Name()) - - if err := data.SetupPersistence(dbFile.Name()); err != nil { - panic(err) - } InitTemporarySingleton(fakeGetStatus) manager = Get() @@ -51,6 +46,8 @@ func TestMain(m *testing.M) { // Because the other tests use `sendEventToWebhooks` with a `WaitGroup` to know when the test completes, // this test ensures that `SendToWebhooks` without a `WaitGroup` doesn't panic. func TestPublicSend(t *testing.T) { + webhookRepository := storage.GetWebhookRepository() + // Send enough events to be sure at least one worker delivers a second event. eventsCount := webhookWorkerPoolSize + 1 @@ -85,6 +82,8 @@ func TestPublicSend(t *testing.T) { // Make sure that events are only sent to interested endpoints. func TestRouting(t *testing.T) { + webhookRepository := storage.GetWebhookRepository() + eventTypes := []models.EventType{models.ChatActionSent, models.UserJoined} calls := map[models.EventType]int{} @@ -140,6 +139,8 @@ func TestRouting(t *testing.T) { // Make sure that events are sent to all interested endpoints. func TestMultiple(t *testing.T) { + webhookRepository := storage.GetWebhookRepository() + const times = 2 var calls uint32 @@ -177,6 +178,8 @@ func TestMultiple(t *testing.T) { // Make sure when a webhook is used its last used timestamp is updated. func TestTimestamps(t *testing.T) { + webhookRepository := storage.GetWebhookRepository() + const tolerance = time.Second start := time.Now() eventTypes := []models.EventType{models.StreamStarted, models.StreamStopped} @@ -258,6 +261,8 @@ func TestTimestamps(t *testing.T) { // Make sure up to the expected number of events can be fired in parallel. func TestParallel(t *testing.T) { + webhookRepository := storage.GetWebhookRepository() + var calls uint32 var wgStart sync.WaitGroup @@ -310,6 +315,8 @@ func TestParallel(t *testing.T) { // Send an event, capture it, and verify that it has the expected payload. func checkPayload(t *testing.T, eventType models.EventType, send func(), expectedJson string) { + webhookRepository := storage.GetWebhookRepository() + eventChannel := make(chan WebhookEvent) // Set up a server. diff --git a/services/webhooks/workerpool.go b/services/webhooks/workerpool.go index 6db1d8848..8ddbaa5a5 100644 --- a/services/webhooks/workerpool.go +++ b/services/webhooks/workerpool.go @@ -10,6 +10,7 @@ import ( log "github.com/sirupsen/logrus" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/storage" ) // webhookWorkerPoolSize defines the number of concurrent HTTP webhook requests. @@ -75,6 +76,8 @@ func (w *LiveWebhookManager) sendWebhook(job Job) error { defer resp.Body.Close() + webhookRepository := storage.GetWebhookRepository() + if err := webhookRepository.SetWebhookAsUsed(job.webhook); err != nil { log.Warnln(err) } diff --git a/services/yp/yp.go b/services/yp/yp.go index 0b7d701f0..105aaac40 100644 --- a/services/yp/yp.go +++ b/services/yp/yp.go @@ -18,12 +18,10 @@ import ( const pingInterval = 4 * time.Minute var ( - getStatus func() models.Status + getStatus func() *models.Status _inErrorState = false ) -var configRepository = configrepository.Get() - // YP is a service for handling listing in the Owncast directory. type YP struct { timer *time.Ticker @@ -42,7 +40,7 @@ type ypPingRequest struct { } // NewYP creates a new instance of the YP service handler. -func NewYP(getStatusFunc func() models.Status) *YP { +func NewYP(getStatusFunc func() *models.Status) *YP { getStatus = getStatusFunc return &YP{} } @@ -63,6 +61,8 @@ func (yp *YP) Stop() { } func (yp *YP) ping() { + configRepository := configrepository.Get() + if !configRepository.GetDirectoryEnabled() { return } diff --git a/static/emoji.go b/static/emoji.go index cc99bda86..01bb9e8c6 100644 --- a/static/emoji.go +++ b/static/emoji.go @@ -1,8 +1,6 @@ package static import ( - "fmt" - "io" "io/fs" "os" "path/filepath" @@ -11,8 +9,6 @@ import ( "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" - "github.com/owncast/owncast/utils" - "github.com/pkg/errors" log "github.com/sirupsen/logrus" ) @@ -88,88 +84,3 @@ func GetEmojiList() []models.CustomEmoji { return emojiData } - -// SetupEmojiDirectory sets up the custom emoji directory by copying all built-in -// emojis if the directory does not yet exist. -func SetupEmojiDirectory() (err error) { - type emojiDirectory struct { - path string - isDir bool - } - - c := config.GetConfig() - - if utils.DoesFileExists(c.CustomEmojiPath) { - return nil - } - - if err = os.MkdirAll(c.CustomEmojiPath, 0o750); err != nil { - return fmt.Errorf("unable to create custom emoji directory: %w", err) - } - - staticFS := GetEmoji() - files := []emojiDirectory{} - - walkFunction := func(path string, d os.DirEntry, err error) error { - if path == "." { - return nil - } - - if d.Name() == "LICENSE.md" { - return nil - } - - files = append(files, emojiDirectory{path: path, isDir: d.IsDir()}) - return nil - } - - if err := fs.WalkDir(staticFS, ".", walkFunction); err != nil { - log.Errorln("unable to fetch emojis: " + err.Error()) - return errors.Wrap(err, "unable to fetch embedded emoji files") - } - - if err != nil { - return fmt.Errorf("unable to read built-in emoji files: %w", err) - } - - // Now copy all built-in emojis to the custom emoji directory - for _, path := range files { - emojiPath := filepath.Join(c.CustomEmojiPath, path.path) - - if path.isDir { - if err := os.Mkdir(emojiPath, 0o700); err != nil { - return errors.Wrap(err, "unable to create emoji directory, check permissions?: "+path.path) - } - continue - } - - memFile, staticOpenErr := staticFS.Open(path.path) - if staticOpenErr != nil { - return errors.Wrap(staticOpenErr, "unable to open emoji file from embedded filesystem") - } - - // nolint:gosec - diskFile, err := os.Create(emojiPath) - if err != nil { - return fmt.Errorf("unable to create custom emoji file on disk: %w", err) - } - - if err != nil { - _ = diskFile.Close() - return fmt.Errorf("unable to open built-in emoji file: %w", err) - } - - if _, err = io.Copy(diskFile, memFile); err != nil { - _ = diskFile.Close() - _ = os.Remove(emojiPath) - return fmt.Errorf("unable to copy built-in emoji file to disk: %w", err) - } - - if err = diskFile.Close(); err != nil { - _ = os.Remove(emojiPath) - return fmt.Errorf("unable to close custom emoji file on disk: %w", err) - } - } - - return nil -} diff --git a/storage/chatrepository/chatrepository.go b/storage/chatrepository/chatrepository.go index 4b0d9948d..2d3916edd 100644 --- a/storage/chatrepository/chatrepository.go +++ b/storage/chatrepository/chatrepository.go @@ -11,6 +11,8 @@ func New(datastore *data.Store) *ChatRepository { datastore: datastore, } + r.startPruner() + return r } @@ -18,7 +20,7 @@ func New(datastore *data.Store) *ChatRepository { var temporaryGlobalInstance *ChatRepository // GetUserRepository will return the user repository. -func GetChatRepository() *ChatRepository { +func Get() *ChatRepository { if temporaryGlobalInstance == nil { i := New(data.GetDatastore()) temporaryGlobalInstance = i diff --git a/core/chat/persistence.go b/storage/chatrepository/persistence.go index ea8d1afb5..a5e9430b1 100644 --- a/core/chat/persistence.go +++ b/storage/chatrepository/persistence.go @@ -1,4 +1,4 @@ -package chat +package chatrepository import ( "context" @@ -6,7 +6,6 @@ import ( "strings" "time" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" "github.com/owncast/owncast/storage/data" log "github.com/sirupsen/logrus" @@ -14,36 +13,17 @@ import ( var _datastore *data.Store -const ( - maxBacklogHours = 2 // Keep backlog max hours worth of messages - maxBacklogNumber = 50 // Return max number of messages in history request -) - -func setupPersistence() { - _datastore = data.GetDatastore() - data.CreateMessagesTable(_datastore.DB) - data.CreateBanIPTable(_datastore.DB) - - chatDataPruner := time.NewTicker(5 * time.Minute) - go func() { - runPruner() - for range chatDataPruner.C { - runPruner() - } - }() -} - // SaveUserMessage will save a single chat event to the messages database. -func SaveUserMessage(event events.UserMessageEvent) { - saveEvent(event.ID, &event.User.ID, event.Body, event.Type, event.HiddenAt, event.Timestamp, nil, nil, nil, nil) +func (cr *ChatRepository) SaveUserMessage(event models.UserMessageEvent) { + cr.SaveEvent(event.ID, &event.User.ID, event.Body, event.Type, event.HiddenAt, event.Timestamp, nil, nil, nil, nil) } -func saveFederatedAction(event events.FediverseEngagementEvent) { - saveEvent(event.ID, nil, event.Body, event.Type, nil, event.Timestamp, event.Image, &event.Link, &event.UserAccountName, nil) +func (cr *ChatRepository) SaveFederatedAction(event models.FediverseEngagementEvent) { + cr.SaveEvent(event.ID, nil, event.Body, event.Type, nil, event.Timestamp, event.Image, &event.Link, &event.UserAccountName, nil) } // nolint: unparam -func saveEvent(id string, userID *string, body string, eventType string, hidden *time.Time, timestamp time.Time, image *string, link *string, title *string, subtitle *string) { +func (cr *ChatRepository) SaveEvent(id string, userID *string, body string, eventType string, hidden *time.Time, timestamp time.Time, image *string, link *string, title *string, subtitle *string) { defer func() { _historyCache = nil }() @@ -74,7 +54,7 @@ func saveEvent(id string, userID *string, body string, eventType string, hidden } } -func makeUserMessageEventFromRowData(row rowData) events.UserMessageEvent { +func (cr *ChatRepository) makeUserMessageEventFromRowData(row rowData) models.UserMessageEvent { scopes := "" if row.userScopes != nil { scopes = *row.userScopes @@ -117,17 +97,17 @@ func makeUserMessageEventFromRowData(row rowData) events.UserMessageEvent { IsBot: isBot, } - message := events.UserMessageEvent{ - Event: events.Event{ + message := models.UserMessageEvent{ + Event: models.Event{ Type: row.eventType, ID: row.id, Timestamp: row.timestamp, }, - UserEvent: events.UserEvent{ + UserEvent: models.UserEvent{ User: &u, HiddenAt: row.hiddenAt, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ Body: row.body, RawBody: row.body, }, @@ -136,14 +116,14 @@ func makeUserMessageEventFromRowData(row rowData) events.UserMessageEvent { return message } -func makeSystemMessageChatEventFromRowData(row rowData) events.SystemMessageEvent { - message := events.SystemMessageEvent{ - Event: events.Event{ +func (cr *ChatRepository) makeSystemMessageChatEventFromRowData(row rowData) models.SystemMessageEvent { + message := models.SystemMessageEvent{ + Event: models.Event{ Type: row.eventType, ID: row.id, Timestamp: row.timestamp, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ Body: row.body, RawBody: row.body, }, @@ -151,14 +131,14 @@ func makeSystemMessageChatEventFromRowData(row rowData) events.SystemMessageEven return message } -func makeActionMessageChatEventFromRowData(row rowData) events.ActionEvent { - message := events.ActionEvent{ - Event: events.Event{ +func (cr *ChatRepository) makeActionMessageChatEventFromRowData(row rowData) models.ActionEvent { + message := models.ActionEvent{ + Event: models.Event{ Type: row.eventType, ID: row.id, Timestamp: row.timestamp, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ Body: row.body, RawBody: row.body, }, @@ -166,14 +146,14 @@ func makeActionMessageChatEventFromRowData(row rowData) events.ActionEvent { return message } -func makeFederatedActionChatEventFromRowData(row rowData) events.FediverseEngagementEvent { - message := events.FediverseEngagementEvent{ - Event: events.Event{ +func (cr *ChatRepository) makeFederatedActionChatEventFromRowData(row rowData) models.FediverseEngagementEvent { + message := models.FediverseEngagementEvent{ + Event: models.Event{ Type: row.eventType, ID: row.id, Timestamp: row.timestamp, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ Body: row.body, RawBody: row.body, }, @@ -208,7 +188,7 @@ type rowData struct { id string } -func getChat(rows *sql.Rows) ([]interface{}, error) { +func (cr *ChatRepository) getChat(rows *sql.Rows) ([]interface{}, error) { history := make([]interface{}, 0) for rows.Next() { @@ -242,18 +222,18 @@ func getChat(rows *sql.Rows) ([]interface{}, error) { var message interface{} switch row.eventType { - case events.MessageSent: - message = makeUserMessageEventFromRowData(row) - case events.SystemMessageSent: - message = makeSystemMessageChatEventFromRowData(row) - case events.ChatActionSent: - message = makeActionMessageChatEventFromRowData(row) - case events.FediverseEngagementFollow: - message = makeFederatedActionChatEventFromRowData(row) - case events.FediverseEngagementLike: - message = makeFederatedActionChatEventFromRowData(row) - case events.FediverseEngagementRepost: - message = makeFederatedActionChatEventFromRowData(row) + case models.MessageSent: + message = cr.makeUserMessageEventFromRowData(row) + case models.SystemMessageSent: + message = cr.makeSystemMessageChatEventFromRowData(row) + case models.ChatActionSent: + message = cr.makeActionMessageChatEventFromRowData(row) + case models.FediverseEngagementFollow: + message = cr.makeFederatedActionChatEventFromRowData(row) + case models.FediverseEngagementLike: + message = cr.makeFederatedActionChatEventFromRowData(row) + case models.FediverseEngagementRepost: + message = cr.makeFederatedActionChatEventFromRowData(row) } history = append(history, message) @@ -265,7 +245,7 @@ func getChat(rows *sql.Rows) ([]interface{}, error) { var _historyCache *[]interface{} // GetChatModerationHistory will return all the chat messages suitable for moderation purposes. -func GetChatModerationHistory() []interface{} { +func (cr *ChatRepository) GetChatModerationHistory() []interface{} { if _historyCache != nil { return *_historyCache } @@ -295,7 +275,7 @@ func GetChatModerationHistory() []interface{} { defer stmt.Close() defer rows.Close() - result, err := getChat(rows) + result, err := cr.getChat(rows) if err != nil { log.Errorln(err) log.Errorln("There is a problem enumerating chat message rows. Please report this:", query) @@ -313,7 +293,7 @@ func GetChatModerationHistory() []interface{} { } // GetChatHistory will return all the chat messages suitable for returning as user-facing chat history. -func GetChatHistory() []interface{} { +func (cr *ChatRepository) GetChatHistory() []interface{} { tx, err := _datastore.DB.Begin() if err != nil { log.Errorln("error fetching chat history", err) @@ -340,7 +320,7 @@ func GetChatHistory() []interface{} { defer stmt.Close() defer rows.Close() - m, err := getChat(rows) + m, err := cr.getChat(rows) if err != nil { log.Errorln(err) log.Errorln("There is a problem enumerating chat message rows. Please report this:", query) @@ -361,20 +341,20 @@ func GetChatHistory() []interface{} { } // GetMessagesFromUser returns chat messages that were sent by a specific user. -func GetMessagesFromUser(userID string) ([]events.UserMessageEvent, error) { +func (cr *ChatRepository) GetMessagesFromUser(userID string) ([]models.UserMessageEvent, error) { query, err := _datastore.GetQueries().GetMessagesFromUser(context.Background(), sql.NullString{String: userID, Valid: true}) if err != nil { return nil, err } - results := make([]events.UserMessageEvent, len(query)) + results := make([]models.UserMessageEvent, len(query)) for i, row := range query { - results[i] = events.UserMessageEvent{ - Event: events.Event{ + results[i] = models.UserMessageEvent{ + Event: models.Event{ Timestamp: row.Timestamp.Time, ID: row.ID, }, - MessageEvent: events.MessageEvent{ + MessageEvent: models.MessageEvent{ Body: row.Body.String, }, } @@ -385,7 +365,7 @@ func GetMessagesFromUser(userID string) ([]events.UserMessageEvent, error) { // SetMessageVisibilityForUserID will bulk change the visibility of messages for a user // and then send out visibility changed events to chat clients. -func SetMessageVisibilityForUserID(userID string, visible bool) error { +func (cr *ChatRepository) SetMessageVisibilityForUserID(userID string, visible bool) ([]string, error) { defer func() { _historyCache = nil }() @@ -393,7 +373,7 @@ func SetMessageVisibilityForUserID(userID string, visible bool) error { tx, err := _datastore.DB.Begin() if err != nil { log.Errorln("error while setting message visibility", err) - return nil + return nil, err } defer tx.Rollback() // nolint @@ -402,13 +382,13 @@ func SetMessageVisibilityForUserID(userID string, visible bool) error { stmt, err := tx.Prepare(query) if err != nil { log.Errorln("error while setting message visibility", err) - return nil + return nil, err } rows, err := stmt.Query(userID) if err != nil { log.Errorln("error while setting message visibility", err) - return nil + return nil, err } defer stmt.Close() @@ -417,31 +397,30 @@ func SetMessageVisibilityForUserID(userID string, visible bool) error { // Get a list of IDs to send to the connected clients to hide ids := make([]string, 0) - messages, err := getChat(rows) + messages, err := cr.getChat(rows) if err != nil { log.Errorln(err) log.Errorln("There is a problem enumerating chat message rows. Please report this:", query) - return nil + return nil, err } if len(messages) == 0 { - return nil + return nil, nil } for _, message := range messages { - ids = append(ids, message.(events.UserMessageEvent).ID) + ids = append(ids, message.(models.UserMessageEvent).ID) } if err = tx.Commit(); err != nil { log.Errorln("error while setting message visibility ", err) - return nil + return nil, nil } - // Tell the clients to hide/show these messages. - return SetMessagesVisibility(ids, visible) + return nil, nil } -func saveMessageVisibility(messageIDs []string, visible bool) error { +func (cr *ChatRepository) SaveMessageVisibility(messageIDs []string, visible bool) error { defer func() { _historyCache = nil }() diff --git a/core/chat/pruner.go b/storage/chatrepository/pruner.go index a9281b3df..0ab357c77 100644 --- a/core/chat/pruner.go +++ b/storage/chatrepository/pruner.go @@ -1,14 +1,33 @@ -package chat +package chatrepository import ( "fmt" + "time" + "github.com/owncast/owncast/storage/data" log "github.com/sirupsen/logrus" ) +const ( + maxBacklogHours = 2 // Keep backlog max hours worth of messages + maxBacklogNumber = 50 // Return max number of messages in history request +) + +func (cr *ChatRepository) startPruner() { + chatDataPruner := time.NewTicker(5 * time.Minute) + go func() { + cr.runPruner() + for range chatDataPruner.C { + cr.runPruner() + } + }() +} + // Only keep recent messages so we don't keep more chat data than needed // for privacy and efficiency reasons. -func runPruner() { +func (cr *ChatRepository) runPruner() { + _datastore := data.GetDatastore() + _datastore.DbLock.Lock() defer _datastore.DbLock.Unlock() diff --git a/storage/configrepository/config.go b/storage/configrepository/config.go index b5ee9ed7b..37409c7b7 100644 --- a/storage/configrepository/config.go +++ b/storage/configrepository/config.go @@ -537,7 +537,7 @@ func (cr *SqlConfigRepository) GetVideoCodec() string { // VerifySettings will perform a sanity check for specific settings values. func (cr *SqlConfigRepository) VerifySettings() error { - c := config.GetConfig() + c := config.Get() if len(cr.GetStreamKeys()) == 0 && c.TemporaryStreamKey == "" { log.Errorln("No stream key set. Streaming is disabled. Please set one via the admin or command line arguments") } diff --git a/storage/configrepository/configrepository.go b/storage/configrepository/configrepository.go index 7d96f4c6d..36879b770 100644 --- a/storage/configrepository/configrepository.go +++ b/storage/configrepository/configrepository.go @@ -119,6 +119,7 @@ type ConfigRepository interface { GetDisableSearchIndexing() bool GetVideoServingEndpoint() string SetVideoServingEndpoint(message string) error + GetDefaultFederationUsername() string } type SqlConfigRepository struct { diff --git a/storage/data/datastore.go b/storage/data/datastore.go index 6997e67dd..f34722525 100644 --- a/storage/data/datastore.go +++ b/storage/data/datastore.go @@ -47,7 +47,7 @@ var temporaryGlobalDatastoreInstance *Store // GetDatastore returns the shared instance of the owncast datastore. func GetDatastore() *Store { if temporaryGlobalDatastoreInstance == nil { - c := config.GetConfig() + c := config.Get() i, err := NewStore(c.DatabaseFilePath) if err != nil { log.Fatal(err) diff --git a/storage/federationrepository/federationrepository.go b/storage/federationrepository/federationrepository.go index d10eea10a..c010494f3 100644 --- a/storage/federationrepository/federationrepository.go +++ b/storage/federationrepository/federationrepository.go @@ -9,9 +9,9 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/resolvers" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/resolvers" "github.com/owncast/owncast/storage/data" "github.com/owncast/owncast/storage/sqlstorage" "github.com/owncast/owncast/utils" @@ -306,6 +306,8 @@ func (f *FederationRepository) GetOutboxPostCount() (int64, error) { func (f *FederationRepository) GetOutbox(limit int, offset int) (vocab.ActivityStreamsOrderedCollection, error) { collection := streams.NewActivityStreamsOrderedCollection() orderedItems := streams.NewActivityStreamsOrderedItemsProperty() + r := resolvers.Get() + rows, err := f.datastore.GetQueries().GetOutboxWithOffset( context.Background(), sqlstorage.GetOutboxWithOffsetParams{Limit: int32(limit), Offset: int32(offset)}, @@ -319,7 +321,7 @@ func (f *FederationRepository) GetOutbox(limit int, offset int) (vocab.ActivityS orderedItems.AppendActivityStreamsCreate(activity) return nil } - if err := resolvers.Resolve(context.Background(), value, createCallback); err != nil { + if err := r.Resolve(context.Background(), value, createCallback); err != nil { return collection, err } } diff --git a/storage/sqlstorage/initialize.go b/storage/sqlstorage/initialize.go index a217ef144..4a5931b88 100644 --- a/storage/sqlstorage/initialize.go +++ b/storage/sqlstorage/initialize.go @@ -91,7 +91,7 @@ func InitializeDatabase(file string, schemaVersion int) (*sql.DB, error) { dbBackupTicker := time.NewTicker(1 * time.Hour) go func() { - c := config.GetConfig() + c := config.Get() backupFile := filepath.Join(c.BackupDirectory, "owncastdb.bak") for range dbBackupTicker.C { utils.Backup(db, backupFile) diff --git a/storage/sqlstorage/migrations.go b/storage/sqlstorage/migrations.go index 7a95a7859..d411a4777 100644 --- a/storage/sqlstorage/migrations.go +++ b/storage/sqlstorage/migrations.go @@ -21,7 +21,7 @@ func NewSqlMigrations(db *sql.DB) *SqlMigrations { } func (m *SqlMigrations) MigrateDatabaseSchema(db *sql.DB, from, to int) error { - c := config.GetConfig() + c := config.Get() log.Printf("Migrating database from version %d to %d", from, to) dbBackupFile := filepath.Join(c.BackupDirectory, fmt.Sprintf("owncast-v%d.bak", from)) diff --git a/storage/userrepository/userrepository.go b/storage/userrepository/userrepository.go index eea8144f3..2d080031d 100644 --- a/storage/userrepository/userrepository.go +++ b/storage/userrepository/userrepository.go @@ -59,17 +59,6 @@ func Get() *SqlUserRepository { return temporaryGlobalInstance } -const ( - // ScopeCanSendChatMessages will allow sending chat messages as itself. - ScopeCanSendChatMessages = "CAN_SEND_MESSAGES" - // ScopeCanSendSystemMessages will allow sending chat messages as the system. - ScopeCanSendSystemMessages = "CAN_SEND_SYSTEM_MESSAGES" - // ScopeHasAdminAccess will allow performing administrative actions on the server. - ScopeHasAdminAccess = "HAS_ADMIN_ACCESS" - - moderatorScopeKey = "MODERATOR" -) - // User represents a single chat user. // SetupUsers will perform the initial initialization of the user package. @@ -285,10 +274,10 @@ func (r *SqlUserRepository) SetUserAsAuthenticated(userID string) error { // SetModerator will add or remove moderator status for a single user by ID. func (r *SqlUserRepository) SetModerator(userID string, isModerator bool) error { if isModerator { - return r.addScopeToUser(userID, moderatorScopeKey) + return r.addScopeToUser(userID, models.ModeratorScopeKey) } - return r.removeScopeFromUser(userID, moderatorScopeKey) + return r.removeScopeFromUser(userID, models.ModeratorScopeKey) } func (r *SqlUserRepository) addScopeToUser(userID string, scope string) error { @@ -401,7 +390,7 @@ func (r *SqlUserRepository) GetModeratorUsers() []*models.User { ORDER BY created_at ) AS token WHERE token.scope = ?` - rows, err := r.datastore.DB.Query(query, moderatorScopeKey) + rows, err := r.datastore.DB.Query(query, models.ModeratorScopeKey) if err != nil { log.Errorln(err) return nil @@ -748,9 +737,9 @@ func (r *SqlUserRepository) makeExternalAPIUsersFromRows(rows *sql.Rows) ([]mode func (r *SqlUserRepository) HasValidScopes(scopes []string) bool { // For a scope to be seen as "valid" it must live in this slice. validAccessTokenScopes := []string{ - ScopeCanSendChatMessages, - ScopeCanSendSystemMessages, - ScopeHasAdminAccess, + models.ScopeCanSendChatMessages, + models.ScopeCanSendSystemMessages, + models.ScopeHasAdminAccess, } for _, scope := range scopes { diff --git a/utils/emojiMigration.go b/utils/emojiMigration.go deleted file mode 100644 index 29700c729..000000000 --- a/utils/emojiMigration.go +++ /dev/null @@ -1,23 +0,0 @@ -package utils - -import ( - "path" - - log "github.com/sirupsen/logrus" -) - -// MigrateCustomEmojiLocations migrates custom emoji from the old location to the new location. -func MigrateCustomEmojiLocations() { - oldLocation := path.Join("webroot", "img", "emoji") - newLocation := path.Join("data", "emoji") - - if !DoesFileExists(oldLocation) { - return - } - - log.Println("Moving custom emoji directory from", oldLocation, "to", newLocation) - - if err := Move(oldLocation, newLocation); err != nil { - log.Errorln("error moving custom emoji directory", err) - } -} diff --git a/video/rtmp/broadcaster.go b/video/rtmp/broadcaster.go index a2d212926..57fa1b32f 100644 --- a/video/rtmp/broadcaster.go +++ b/video/rtmp/broadcaster.go @@ -30,5 +30,5 @@ func setCurrentBroadcasterInfo(t flvio.Tag, remoteAddr string) { }, } - _setBroadcaster(broadcaster) + _setBroadcaster(&broadcaster) } diff --git a/video/rtmp/rtmp.go b/video/rtmp/rtmp.go index fe93b8706..4f8388066 100644 --- a/video/rtmp/rtmp.go +++ b/video/rtmp/rtmp.go @@ -25,16 +25,16 @@ var ( var ( _setStreamAsConnected func(*io.PipeReader) - _setBroadcaster func(models.Broadcaster) + _setBroadcaster func(*models.Broadcaster) ) -var configRepository = configrepository.Get() - // Start starts the rtmp service, listening on specified RTMP port. -func Start(setStreamAsConnected func(*io.PipeReader), setBroadcaster func(models.Broadcaster)) { +func Start(setStreamAsConnected func(*io.PipeReader), setBroadcaster func(*models.Broadcaster)) { _setStreamAsConnected = setStreamAsConnected _setBroadcaster = setBroadcaster + configRepository := configrepository.Get() + port := configRepository.GetRTMPPortNumber() s := rtmp.NewServer() var lis net.Listener @@ -80,10 +80,12 @@ func HandleConn(c *rtmp.Conn, nc net.Conn) { return } + configRepository := configrepository.Get() + accessGranted := false validStreamingKeys := configRepository.GetStreamKeys() - configservice := config.GetConfig() + configservice := config.Get() // If a stream key override was specified then use that instead. if configservice.TemporaryStreamKey != "" { diff --git a/video/state/offline.go b/video/state/offline.go new file mode 100644 index 000000000..31cada7b1 --- /dev/null +++ b/video/state/offline.go @@ -0,0 +1,123 @@ +package state + +import ( + "os" + "path" + "path/filepath" + + "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/utils" + "github.com/owncast/owncast/video/transcoder" + log "github.com/sirupsen/logrus" +) + +// transitionToOfflineVideoStreamContent will overwrite the current stream with the +// offline video stream state only. No live stream HLS segments will continue to be +// referenced. +func transitionToOfflineVideoStreamContent() { + log.Traceln("Firing transcoder with offline stream state") + + _transcoder := transcoder.NewTranscoder() + _transcoder.SetIdentifier("offline") + _transcoder.SetLatencyLevel(models.GetLatencyLevel(4)) + _transcoder.SetIsEvent(true) + + offlineFilePath, err := saveOfflineClipToDisk("offline.ts") + if err != nil { + log.Fatalln("unable to save offline clip:", err) + } + + _transcoder.SetInput(offlineFilePath) + go _transcoder.Start(false) + + configRepository := configrepository.Get() + + // Copy the logo to be the thumbnail + logo := configRepository.GetLogoPath() + c := config.Get() + dst := filepath.Join(c.TempDir, "thumbnail.jpg") + if err = utils.Copy(filepath.Join("data", logo), dst); err != nil { + log.Warnln(err) + } + + // Delete the preview Gif + _ = os.Remove(path.Join(config.DataDirectory, "preview.gif")) +} + +func createInitialOfflineState() error { + transitionToOfflineVideoStreamContent() + + return nil +} + +// SetStreamAsDisconnected sets the stream as disconnected. +func SetStreamAsDisconnected() { + _ = chat.SendSystemAction("The stream is ending.", true) + + now := utils.NullTime{Time: time.Now(), Valid: true} + if _onlineTimerCancelFunc != nil { + _onlineTimerCancelFunc() + } + + _stats.StreamConnected = false + _stats.LastDisconnectTime = &now + _stats.LastConnectTime = nil + _broadcaster = nil + + offlineFilename := "offline.ts" + + offlineFilePath, err := saveOfflineClipToDisk(offlineFilename) + if err != nil { + log.Errorln(err) + return + } + + transcoder.StopThumbnailGenerator() + rtmp.Disconnect() + + if _yp != nil { + _yp.Stop() + } + + // If there is no current broadcast available the previous stream + // likely failed for some reason. Don't try to append to it. + // Just transition to offline. + if _currentBroadcast == nil { + stopOnlineCleanupTimer() + transitionToOfflineVideoStreamContent() + log.Errorln("unexpected nil _currentBroadcast") + return + } + + for index := range _currentBroadcast.OutputSettings { + makeVariantIndexOffline(index, offlineFilePath, offlineFilename) + } + + StartOfflineCleanupTimer() + stopOnlineCleanupTimer() + saveStats() + + webhookManager := webhooks.Get() + go webhookManager.SendStreamStatusEvent(models.StreamStopped) +} + +// StartOfflineCleanupTimer will fire a cleanup after n minutes being disconnected. +func StartOfflineCleanupTimer() { + _offlineCleanupTimer = time.NewTimer(5 * time.Minute) + go func() { + for range _offlineCleanupTimer.C { + // Set video to offline state + resetDirectories() + transitionToOfflineVideoStreamContent() + } + }() +} + +// StopOfflineCleanupTimer will stop the previous cleanup timer. +func StopOfflineCleanupTimer() { + if _offlineCleanupTimer != nil { + _offlineCleanupTimer.Stop() + } +} diff --git a/video/state/online.go b/video/state/online.go new file mode 100644 index 000000000..1b37887e2 --- /dev/null +++ b/video/state/online.go @@ -0,0 +1,124 @@ +package state + +import ( + "context" + "io" + "time" + + "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation/outbox" + "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/services/notifications" + "github.com/owncast/owncast/services/webhooks" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/data" + "github.com/owncast/owncast/utils" + "github.com/owncast/owncast/video/transcoder" + log "github.com/sirupsen/logrus" +) + +// setStreamAsConnected sets the stream as connected. +func (vs *VideoState) setStreamAsConnected(rtmpOut *io.PipeReader) { + now := utils.NullTime{Time: time.Now(), Valid: true} + vs.status.StreamConnected = true + vs.status.Status.LastDisconnectTime = nil + vs.status.Status.LastConnectTime = &now + vs.status.SessionMaxViewerCount = 0 + + configRepository := configrepository.Get() + + vs.status.SetCurrentBroadcast(&models.CurrentBroadcast{ + LatencyLevel: configRepository.GetStreamLatencyLevel(), + OutputSettings: configRepository.GetStreamOutputVariants(), + }) + + StopOfflineCleanupTimer() + startOnlineCleanupTimer() + + if _yp != nil { + go _yp.Start() + } + + c := config.Get() + segmentPath := c.HLSStoragePath + + if err := setupStorage(); err != nil { + log.Fatalln("failed to setup the storage", err) + } + + go func() { + vs.transcoder = transcoder.NewTranscoder() + vs.transcoder.TranscoderCompleted = func(error) { + SetStreamAsDisconnected() + vs.transcoder = nil + vs.status.SetCurrentBroadcast(nil) + } + vs.transcoder.SetStdin(rtmpOut) + vs.transcoder.Start(true) + }() + + webhookManager := webhooks.Get() + go webhookManager.SendStreamStatusEvent(models.StreamStarted) + transcoder.StartThumbnailGenerator(segmentPath, configRepository.FindHighestVideoQualityIndex(vs.status.GetCurrentBroadcast().OutputSettings)) + + _ = vs.chatService.SendSystemAction("Stay tuned, the stream is **starting**!", true) + vs.chatService.SendAllWelcomeMessage() + + // Send delayed notification messages. + _onlineTimerCancelFunc = startLiveStreamNotificationsTimer() +} + +func startOnlineCleanupTimer() { + _onlineCleanupTicker = time.NewTicker(1 * time.Minute) + go func() { + for range _onlineCleanupTicker.C { + if err := _storage.Cleanup(); err != nil { + log.Errorln(err) + } + } + }() +} + +func stopOnlineCleanupTimer() { + if _onlineCleanupTicker != nil { + _onlineCleanupTicker.Stop() + } +} + +func startLiveStreamNotificationsTimer() context.CancelFunc { + // Send delayed notification messages. + c, cancelFunc := context.WithCancel(context.Background()) + _onlineTimerCancelFunc = cancelFunc + go func(c context.Context) { + select { + case <-time.After(time.Minute * 2.0): + if _lastNotified != nil && time.Since(*_lastNotified) < 10*time.Minute { + return + } + + configRepository := configrepository.Get() + + // Send Fediverse message. + if configRepository.GetFederationEnabled() { + ob := outbox.Get() + log.Traceln("Sending Federated Go Live message.") + if err := ob.SendLive(); err != nil { + log.Errorln(err) + } + } + + // Send notification to those who have registered for them. + if notifier, err := notifications.New(data.GetDatastore()); err != nil { + log.Errorln(err) + } else { + notifier.Notify() + } + + now := time.Now() + _lastNotified = &now + case <-c.Done(): + } + }(c) + + return cancelFunc +} diff --git a/video/state/state.go b/video/state/state.go new file mode 100644 index 000000000..baae26854 --- /dev/null +++ b/video/state/state.go @@ -0,0 +1,35 @@ +package state + +import ( + "github.com/owncast/owncast/services/chat" + "github.com/owncast/owncast/services/metrics" + "github.com/owncast/owncast/services/status" + "github.com/owncast/owncast/utils" + "github.com/owncast/owncast/video/transcoder" +) + +type VideoState struct { + transcoder *transcoder.Transcoder + metrics *metrics.Metrics + status *status.Status + lastNotified utils.NullTime + chatService *chat.Chat +} + +func New() *VideoState { + return &VideoState{ + transcoder: transcoder.Get(), + metrics: metrics.Get(), + status: status.Get(), + } +} + +var temporaryGlobalInstance *VideoState + +func Get() *VideoState { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = New() + } + + return temporaryGlobalInstance +} diff --git a/video/storageproviders/local.go b/video/storageproviders/local.go index 361d21d31..986e9a2de 100644 --- a/video/storageproviders/local.go +++ b/video/storageproviders/local.go @@ -9,6 +9,7 @@ import ( log "github.com/sirupsen/logrus" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" ) // LocalStorage represents an instance of the local storage provider for HLS video. @@ -52,10 +53,12 @@ func (s *LocalStorage) Save(filePath string, retryCount int) (string, error) { } func (s *LocalStorage) Cleanup() error { + configRepository := configrepository.Get() + // Determine how many files we should keep on disk maxNumber := configRepository.GetStreamLatencyLevel().SegmentCount buffer := 10 - c := config.GetConfig() + c := config.Get() baseDirectory := c.HLSStoragePath files, err := getAllFilesRecursive(baseDirectory) diff --git a/video/storageproviders/rewriteLocalPlaylist.go b/video/storageproviders/rewriteLocalPlaylist.go index 90f17b63c..739c6afdb 100644 --- a/video/storageproviders/rewriteLocalPlaylist.go +++ b/video/storageproviders/rewriteLocalPlaylist.go @@ -34,7 +34,7 @@ func rewriteRemotePlaylist(localFilePath, remoteServingEndpoint, pathPrefix stri } item.URI = remoteServingEndpoint + filepath.Join(finalPath, item.URI) } - c := config.GetConfig() + c := config.Get() publicPath := filepath.Join(c.HLSStoragePath, filepath.Base(localFilePath)) newPlaylist := p.String() diff --git a/video/storageproviders/s3Storage.go b/video/storageproviders/s3Storage.go index afa72e2c9..5df6687ab 100644 --- a/video/storageproviders/s3Storage.go +++ b/video/storageproviders/s3Storage.go @@ -47,8 +47,6 @@ type S3Storage struct { uploader *s3manager.Uploader } -var configRepository = configrepository.Get() - // NewS3Storage returns a new S3Storage instance. func NewS3Storage() *S3Storage { return &S3Storage{ @@ -59,6 +57,7 @@ func NewS3Storage() *S3Storage { // Setup sets up the s3 storage for saving the video to s3. func (s *S3Storage) Setup() error { log.Trace("Setting up S3 for external storage of video...") + configRepository := configrepository.Get() s3Config := configRepository.GetS3Config() customVideoServingEndpoint := configRepository.GetVideoServingEndpoint() @@ -102,6 +101,8 @@ func (s *S3Storage) SegmentWritten(localFilePath string) { // Warn the user about long-running save operations if averagePerformance != 0 { + configRepository := configrepository.Get() + if averagePerformance > float64(configRepository.GetStreamLatencyLevel().SecondsPerSegment)*0.9 { log.Warnln("Possible slow uploads: average upload S3 save duration", averagePerformance, "s. troubleshoot this issue by visiting https://owncast.online/docs/troubleshooting/") } @@ -151,7 +152,7 @@ func (s *S3Storage) Save(filePath string, retryCount int) (string, error) { } defer file.Close() - c := config.GetConfig() + c := config.Get() // Convert the local path to the variant/file path by stripping the local storage location. normalizedPath := strings.TrimPrefix(filePath, c.HLSStoragePath) @@ -210,6 +211,8 @@ func (s *S3Storage) Save(filePath string, retryCount int) (string, error) { } func (s *S3Storage) Cleanup() error { + configRepository := configrepository.Get() + // Determine how many files we should keep on S3 storage maxNumber := configRepository.GetStreamLatencyLevel().SegmentCount buffer := 20 diff --git a/video/transcoder/fileWriterReceiverService.go b/video/transcoder/fileWriterReceiverService.go index 74fa96ca7..cb7e224c5 100644 --- a/video/transcoder/fileWriterReceiverService.go +++ b/video/transcoder/fileWriterReceiverService.go @@ -42,7 +42,7 @@ func (s *FileWriterReceiverService) SetupFileWriterReceiverService(callbacks Fil log.Fatalln("Unable to start internal video writing service", err) } - c := config.GetConfig() + c := config.Get() listenerPort := strings.Split(listener.Addr().String(), ":")[1] c.InternalHLSListenerPort = listenerPort log.Traceln("Transcoder response service listening on: " + listenerPort) @@ -60,7 +60,7 @@ func (s *FileWriterReceiverService) uploadHandler(w http.ResponseWriter, r *http return } - c := config.GetConfig() + c := config.Get() path := r.URL.Path writePath := filepath.Join(c.HLSStoragePath, path) f, err := os.Create(writePath) //nolint: gosec diff --git a/video/transcoder/thumbnailGenerator.go b/video/transcoder/thumbnailGenerator.go index 2d7795a0c..cb1ea52cd 100644 --- a/video/transcoder/thumbnailGenerator.go +++ b/video/transcoder/thumbnailGenerator.go @@ -15,10 +15,7 @@ import ( "github.com/owncast/owncast/utils" ) -var ( - _timer *time.Ticker - configRepository = configrepository.Get() -) +var _timer *time.Ticker // StopThumbnailGenerator will stop the periodic generating of a thumbnail from video. func StopThumbnailGenerator() { @@ -51,7 +48,7 @@ func StartThumbnailGenerator(chunkPath string, variantIndex int) { } func fireThumbnailGenerator(segmentPath string, variantIndex int) error { - c := config.GetConfig() + c := config.Get() // JPG takes less time to encode than PNG outputFile := path.Join(c.TempDir, "thumbnail.jpg") @@ -90,6 +87,8 @@ func fireThumbnailGenerator(segmentPath string, variantIndex int) error { return nil } + configRepository := configrepository.Get() + mostRecentFile := path.Join(framePath, names[0]) ffmpegPath := utils.ValidatedFfmpegPath(configRepository.GetFfMpegPath()) outputFileTemp := path.Join(c.TempDir, "tempthumbnail.jpg") @@ -121,7 +120,8 @@ func fireThumbnailGenerator(segmentPath string, variantIndex int) error { } func makeAnimatedGifPreview(sourceFile string, outputFile string) { - c := config.GetConfig() + configRepository := configrepository.Get() + c := config.Get() ffmpegPath := utils.ValidatedFfmpegPath(configRepository.GetFfMpegPath()) outputFileTemp := path.Join(c.TempDir, "temppreview.gif") diff --git a/video/transcoder/transcoder.go b/video/transcoder/transcoder.go index d59d3d1ab..8f8fb3b40 100644 --- a/video/transcoder/transcoder.go +++ b/video/transcoder/transcoder.go @@ -11,9 +11,9 @@ import ( log "github.com/sirupsen/logrus" "github.com/teris-io/shortid" - "github.com/owncast/owncast/logging" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/utils" ) @@ -119,7 +119,7 @@ func (t *Transcoder) Start(shouldLog bool) { } createVariantDirectories() - c := config.GetConfig() + c := config.Get() if c.EnableDebugFeatures { log.Println(command) @@ -137,7 +137,7 @@ func (t *Transcoder) Start(shouldLog bool) { } if err := _commandExec.Start(); err != nil { - log.Errorln("Transcoder error. See", logging.GetTranscoderLogFilePath(), "for full output to debug.") + log.Errorln("Transcoder error. See", c.GetTranscoderLogFilePath(), "for full output to debug.") log.Panicln(err, command) } @@ -155,7 +155,7 @@ func (t *Transcoder) Start(shouldLog bool) { } if err != nil { - log.Errorln("transcoding error. look at", logging.GetTranscoderLogFilePath(), "to help debug. your copy of ffmpeg may not support your selected codec of", t.codec.Name(), "https://owncast.online/docs/codecs/") + log.Errorln("transcoding error. look at", c.GetTranscoderLogFilePath(), "to help debug. your copy of ffmpeg may not support your selected codec of", t.codec.Name(), "https://owncast.online/docs/codecs/") } } @@ -198,8 +198,11 @@ func (t *Transcoder) getString() string { if len(hlsOptionFlags) > 0 { hlsOptionsString = "-hls_flags " + strings.Join(hlsOptionFlags, "+") } + + c := config.Get() + ffmpegFlags := []string{ - fmt.Sprintf(`FFREPORT=file="%s":level=32`, logging.GetTranscoderLogFilePath()), + fmt.Sprintf(`FFREPORT=file="%s":level=32`, c.GetTranscoderLogFilePath()), t.ffmpegPath, "-hide_banner", "-loglevel warning", @@ -272,10 +275,22 @@ func getVariantFromConfigQuality(quality models.StreamOutputVariant, index int) return variant } +var temporaryGlobalInstance *Transcoder + +func Get() *Transcoder { + if temporaryGlobalInstance == nil { + temporaryGlobalInstance = NewTranscoder() + } + + return temporaryGlobalInstance +} + // NewTranscoder will return a new Transcoder, populated by the config. func NewTranscoder() *Transcoder { + configRepository := configrepository.Get() + ffmpegPath := utils.ValidatedFfmpegPath(configRepository.GetFfMpegPath()) - c := config.GetConfig() + c := config.Get() transcoder := new(Transcoder) transcoder.ffmpegPath = ffmpegPath diff --git a/video/transcoder/utils.go b/video/transcoder/utils.go index 936e3a0f8..d804b8adb 100644 --- a/video/transcoder/utils.go +++ b/video/transcoder/utils.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/utils" log "github.com/sirupsen/logrus" ) @@ -96,11 +97,13 @@ func handleTranscoderMessage(message string) { } func createVariantDirectories() { - c := config.GetConfig() + c := config.Get() // Create private hls data dirs utils.CleanupDirectory(c.HLSStoragePath) + configRepository := configrepository.Get() + if len(configRepository.GetStreamOutputVariants()) != 0 { for index := range configRepository.GetStreamOutputVariants() { if err := os.MkdirAll(path.Join(c.HLSStoragePath, strconv.Itoa(index)), 0o750); err != nil { diff --git a/web/style-definitions/config.js b/web/style-definitions/config.js index 7ab0ddf20..f9d52ce50 100644 --- a/web/style-definitions/config.js +++ b/web/style-definitions/config.js @@ -41,15 +41,40 @@ module.exports = { }, ], }, - less: { - transformGroup: 'less', + // less: { + // transformGroup: 'less', + // buildPath: 'build/', + // files: [ + // { + // destination: 'variables.less', + // format: 'less/variables', + // options: { + // fileHeader: 'myCustomHeader', + // }, + // }, + // ], + // }, + 'ios-swift': { + transforms: [ + 'attribute/cti', + 'name/cti/camel', + 'color/UIColorSwift', + 'content/swift/literal', + 'size/swift/remToCGFloat', + 'font/swift/literal', + ], buildPath: 'build/', files: [ { - destination: 'variables.less', - format: 'less/variables', + destination: 'Assets.swift', + format: 'ios-swift/enum.swift', + className: 'Assets', + filter: token => { + return token.attributes.category === 'asset'; + }, options: { - fileHeader: 'myCustomHeader', + outputReferences: true, + showFileHeader: false, }, }, ], diff --git a/webserver/handlers/adminApiChatHandlers.go b/webserver/handlers/adminApiChatHandlers.go index a442f0910..53fdb4088 100644 --- a/webserver/handlers/adminApiChatHandlers.go +++ b/webserver/handlers/adminApiChatHandlers.go @@ -9,8 +9,6 @@ import ( "net/http" "strconv" - "github.com/owncast/owncast/core/chat" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/core/data" "github.com/owncast/owncast/models" "github.com/owncast/owncast/storage/chatrepository" @@ -23,7 +21,7 @@ import ( ) var ( - chatRepository = chatrepository.GetChatRepository() + chatRepository = chatrepository.Get() userRepository = userrepository.Get() ) @@ -53,7 +51,7 @@ func (h *Handlers) UpdateMessageVisibility(w http.ResponseWriter, r *http.Reques return } - if err := chat.SetMessagesVisibility(request.IDArray, request.Visible); err != nil { + if err := h.chatService.SetMessagesVisibility(request.IDArray, request.Visible); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -148,16 +146,18 @@ func (h *Handlers) UpdateUserEnabled(w http.ResponseWriter, r *http.Request) { // Hide/show the user's chat messages if disabling. // Leave hidden messages hidden to be safe. if !request.Enabled { - if err := chat.SetMessageVisibilityForUserID(request.UserID, request.Enabled); err != nil { + + if err := h.chatService.SetMessageVisibilityForUserID(request.UserID, request.Enabled); err != nil { log.Errorln("error changing user messages visibility", err) responses.WriteSimpleResponse(w, false, err.Error()) return } + h.chatService.SetMessagesVisibility(ids, visible) } // Forcefully disconnect the user from the chat if !request.Enabled { - clients, err := chat.GetClientsForUser(request.UserID) + clients, err := h.chatService.GetClientsForUser(request.UserID) if len(clients) == 0 { // Nothing to do return @@ -169,9 +169,9 @@ func (h *Handlers) UpdateUserEnabled(w http.ResponseWriter, r *http.Request) { return } - chat.DisconnectClients(clients) + h.chatService.DisconnectClients(clients) disconnectedUser := userRepository.GetUserByID(request.UserID) - _ = chat.SendSystemAction(fmt.Sprintf("**%s** has been removed from chat.", disconnectedUser.DisplayName), true) + _ = h.chatService.SendSystemAction(fmt.Sprintf("**%s** has been removed from chat.", disconnectedUser.DisplayName), true) localIP4Address := "127.0.0.1" localIP6Address := "::1" @@ -226,7 +226,7 @@ func (h *Handlers) UpdateUserModerator(w http.ResponseWriter, r *http.Request) { } // Update the clients for this user to know about the moderator access change. - if err := chat.SendConnectedClientInfoToUser(req.UserID); err != nil { + if err := h.chatService.SendConnectedClientInfoToUser(req.UserID); err != nil { log.Debugln(err) } @@ -245,7 +245,7 @@ func (h *Handlers) GetModerators(w http.ResponseWriter, r *http.Request) { func (h *Handlers) GetAdminChatMessages(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - messages := chat.GetChatModerationHistory() + messages := h.chatRepository.GetChatModerationHistory() responses.WriteResponse(w, messages) } @@ -253,13 +253,13 @@ func (h *Handlers) GetAdminChatMessages(w http.ResponseWriter, r *http.Request) func (h *Handlers) SendSystemMessage(integration models.ExternalAPIUser, w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - var message events.SystemMessageEvent + var message models.SystemMessageEvent if err := json.NewDecoder(r.Body).Decode(&message); err != nil { responses.InternalErrorHandler(w, err) return } - if err := chat.SendSystemMessage(message.Body, false); err != nil { + if err := h.chatService.SendSystemMessage(message.Body, false); err != nil { responses.BadRequestHandler(w, err) } @@ -281,13 +281,13 @@ func (h *Handlers) SendSystemMessageToConnectedClient(integration models.Externa return } - var message events.SystemMessageEvent + var message models.SystemMessageEvent if err := json.NewDecoder(r.Body).Decode(&message); err != nil { responses.InternalErrorHandler(w, err) return } - chat.SendSystemMessageToClient(uint(clientIDNumeric), message.Body) + h.chatService.SendSystemMessageToClient(uint(clientIDNumeric), message.Body) responses.WriteSimpleResponse(w, true, "sent") } @@ -308,7 +308,7 @@ func (h *Handlers) SendIntegrationChatMessage(integration models.ExternalAPIUser return } - var event events.UserMessageEvent + var event models.UserMessageEvent if err := json.NewDecoder(r.Body).Decode(&event); err != nil { responses.InternalErrorHandler(w, err) return @@ -330,12 +330,12 @@ func (h *Handlers) SendIntegrationChatMessage(integration models.ExternalAPIUser IsBot: true, } - if err := chat.Broadcast(&event); err != nil { + if err := h.chatService.Broadcast(&event); err != nil { responses.BadRequestHandler(w, err) return } - chat.SaveUserMessage(event) + h.chatRepository.SaveUserMessage(event) responses.WriteSimpleResponse(w, true, "sent") } @@ -344,7 +344,7 @@ func (h *Handlers) SendIntegrationChatMessage(integration models.ExternalAPIUser func (h *Handlers) SendChatAction(integration models.ExternalAPIUser, w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - var message events.SystemActionEvent + var message models.SystemActionEvent if err := json.NewDecoder(r.Body).Decode(&message); err != nil { responses.InternalErrorHandler(w, err) return @@ -353,7 +353,7 @@ func (h *Handlers) SendChatAction(integration models.ExternalAPIUser, w http.Res message.SetDefaults() message.RenderBody() - if err := chat.SendSystemAction(message.Body, false); err != nil { + if err := h.chatService.SendSystemAction(message.Body, false); err != nil { responses.BadRequestHandler(w, err) return } diff --git a/webserver/handlers/adminApiConfigHandlers.go b/webserver/handlers/adminApiConfigHandlers.go index 8e03f8a93..581ca3478 100644 --- a/webserver/handlers/adminApiConfigHandlers.go +++ b/webserver/handlers/adminApiConfigHandlers.go @@ -9,9 +9,8 @@ import ( "path/filepath" "strings" - "github.com/owncast/owncast/activitypub/outbox" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation/outbox" "github.com/owncast/owncast/services/webhooks" "github.com/owncast/owncast/utils" "github.com/owncast/owncast/webserver/requests" @@ -41,8 +40,10 @@ func (h *Handlers) SetTags(w http.ResponseWriter, r *http.Request) { return } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -68,7 +69,7 @@ func (h *Handlers) SetStreamTitle(w http.ResponseWriter, r *http.Request) { return } if value != "" { - sendSystemChatAction(fmt.Sprintf("Stream title changed to **%s**", value), true) + h.sendSystemChatAction(fmt.Sprintf("Stream title changed to **%s**", value), true) webhookManager := webhooks.Get() go webhookManager.SendStreamStatusEvent(models.StreamTitleUpdated) } @@ -80,8 +81,8 @@ func (h *Handlers) ExternalSetStreamTitle(integration models.ExternalAPIUser, w h.SetStreamTitle(w, r) } -func sendSystemChatAction(messageText string, ephemeral bool) { - if err := chat.SendSystemAction(messageText, ephemeral); err != nil { +func (h *Handlers) sendSystemChatAction(messageText string, ephemeral bool) { + if err := h.chatService.SendSystemAction(messageText, ephemeral); err != nil { log.Errorln(err) } } @@ -102,8 +103,10 @@ func (h *Handlers) SetServerName(w http.ResponseWriter, r *http.Request) { return } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -127,8 +130,10 @@ func (h *Handlers) SetServerSummary(w http.ResponseWriter, r *http.Request) { return } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -249,8 +254,10 @@ func (h *Handlers) SetLogo(w http.ResponseWriter, r *http.Request) { log.Error("Error saving logo uniqueness string: ", err) } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -560,8 +567,10 @@ func (h *Handlers) SetSocialHandles(w http.ResponseWriter, r *http.Request) { return } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } diff --git a/webserver/handlers/adminApiConnectedClients.go b/webserver/handlers/adminApiConnectedClients.go index bce672b32..38be09639 100644 --- a/webserver/handlers/adminApiConnectedClients.go +++ b/webserver/handlers/adminApiConnectedClients.go @@ -4,14 +4,13 @@ import ( "encoding/json" "net/http" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" "github.com/owncast/owncast/webserver/responses" ) // GetConnectedChatClients returns currently connected clients. func (h *Handlers) GetConnectedChatClients(w http.ResponseWriter, r *http.Request) { - clients := chat.GetClients() + clients := h.chatService.GetClients() w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(clients); err != nil { diff --git a/webserver/handlers/adminApiEmojiConfig.go b/webserver/handlers/adminApiEmojiConfig.go index 21fea849b..2a6562f4a 100644 --- a/webserver/handlers/adminApiEmojiConfig.go +++ b/webserver/handlers/adminApiEmojiConfig.go @@ -38,7 +38,7 @@ func (h *Handlers) UploadCustomEmoji(w http.ResponseWriter, r *http.Request) { } // Prevent path traversal attacks - c := config.GetConfig() + c := config.Get() emojiFileName := filepath.Base(emoji.Name) targetPath := filepath.Join(c.CustomEmojiPath, emojiFileName) @@ -78,7 +78,7 @@ func (h *Handlers) DeleteCustomEmoji(w http.ResponseWriter, r *http.Request) { return } - c := config.GetConfig() + c := config.Get() targetPath := filepath.Join(c.CustomEmojiPath, emoji.Name) if err := os.Remove(targetPath); err != nil { diff --git a/webserver/handlers/adminApiFederationConfig.go b/webserver/handlers/adminApiFederationConfig.go index ca0e3ce38..c2452899e 100644 --- a/webserver/handlers/adminApiFederationConfig.go +++ b/webserver/handlers/adminApiFederationConfig.go @@ -3,9 +3,9 @@ package handlers import ( "net/http" - "github.com/owncast/owncast/activitypub" - "github.com/owncast/owncast/activitypub/outbox" - "github.com/owncast/owncast/activitypub/persistence" + "github.com/owncast/owncast/services/apfederation" + "github.com/owncast/owncast/services/apfederation/outbox" + "github.com/owncast/owncast/storage/federationrepository" "github.com/owncast/owncast/webserver/requests" "github.com/owncast/owncast/webserver/responses" ) @@ -27,7 +27,9 @@ func (h *Handlers) SendFederatedMessage(w http.ResponseWriter, r *http.Request) return } - if err := activitypub.SendPublicFederatedMessage(message); err != nil { + ap := apfederation.Get() + + if err := ap.SendPublicFederatedMessage(message); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -69,8 +71,10 @@ func (h *Handlers) SetFederationActivityPrivate(w http.ResponseWriter, r *http.R return } + ob := outbox.Get() + // Update Fediverse followers about this change. - if err := outbox.UpdateFollowersWithAccountUpdates(); err != nil { + if err := ob.UpdateFollowersWithAccountUpdates(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -164,7 +168,9 @@ func (h *Handlers) SetFederationBlockDomains(w http.ResponseWriter, r *http.Requ func (h *Handlers) GetFederatedActions(page int, pageSize int, w http.ResponseWriter, r *http.Request) { offset := pageSize * page - activities, total, err := persistence.GetInboundActivities(pageSize, offset) + federationRepository := federationrepository.Get() + + activities, total, err := federationRepository.GetInboundActivities(pageSize, offset) if err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return diff --git a/webserver/handlers/adminApiFollowers.go b/webserver/handlers/adminApiFollowers.go index 576559549..e64794952 100644 --- a/webserver/handlers/adminApiFollowers.go +++ b/webserver/handlers/adminApiFollowers.go @@ -4,8 +4,8 @@ import ( "encoding/json" "net/http" - "github.com/owncast/owncast/activitypub/persistence" - aprequests "github.com/owncast/owncast/activitypub/requests" + "github.com/owncast/owncast/services/apfederation/outbox" + "github.com/owncast/owncast/storage/federationrepository" "github.com/owncast/owncast/webserver/requests" "github.com/owncast/owncast/webserver/responses" ) @@ -28,29 +28,33 @@ func (h *Handlers) ApproveFollower(w http.ResponseWriter, r *http.Request) { return } + federationRepository := federationrepository.Get() + if approval.Approved { // Approve a follower - if err := persistence.ApprovePreviousFollowRequest(approval.ActorIRI); err != nil { + if err := federationRepository.ApprovePreviousFollowRequest(approval.ActorIRI); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } localAccountName := configRepository.GetDefaultFederationUsername() - followRequest, err := persistence.GetFollower(approval.ActorIRI) + followRequest, err := federationRepository.GetFollower(approval.ActorIRI) if err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } + ob := outbox.Get() + // Send the approval to the follow requestor. - if err := aprequests.SendFollowAccept(followRequest.Inbox, followRequest.RequestObject, localAccountName); err != nil { + if err := ob.SendFollowAccept(followRequest.Inbox, followRequest.RequestObject, localAccountName); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } } else { // Remove/block a follower - if err := persistence.BlockOrRejectFollower(approval.ActorIRI); err != nil { + if err := federationRepository.BlockOrRejectFollower(approval.ActorIRI); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } @@ -61,7 +65,8 @@ func (h *Handlers) ApproveFollower(w http.ResponseWriter, r *http.Request) { // GetPendingFollowRequests will return a list of pending follow requests. func (h *Handlers) GetPendingFollowRequests(w http.ResponseWriter, r *http.Request) { - requests, err := persistence.GetPendingFollowRequests() + federationRepository := federationrepository.Get() + requests, err := federationRepository.GetPendingFollowRequests() if err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return @@ -72,7 +77,8 @@ func (h *Handlers) GetPendingFollowRequests(w http.ResponseWriter, r *http.Reque // GetBlockedAndRejectedFollowers will return blocked and rejected followers. func (h *Handlers) GetBlockedAndRejectedFollowers(w http.ResponseWriter, r *http.Request) { - rejections, err := persistence.GetBlockedAndRejectedFollowers() + federationRepository := federationrepository.Get() + rejections, err := federationRepository.GetBlockedAndRejectedFollowers() if err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return diff --git a/webserver/handlers/adminApiHardwareStats.go b/webserver/handlers/adminApiHardwareStats.go index c6a806c46..41ca07e1c 100644 --- a/webserver/handlers/adminApiHardwareStats.go +++ b/webserver/handlers/adminApiHardwareStats.go @@ -10,7 +10,7 @@ import ( // GetHardwareStats will return hardware utilization over time. func (h *Handlers) GetHardwareStats(w http.ResponseWriter, r *http.Request) { - m := metrics.GetMetrics() + m := metrics.Get() w.Header().Set("Content-Type", "application/json") err := json.NewEncoder(w).Encode(m) diff --git a/webserver/handlers/adminApiServerConfig.go b/webserver/handlers/adminApiServerConfig.go index d907b9a15..df8a200a8 100644 --- a/webserver/handlers/adminApiServerConfig.go +++ b/webserver/handlers/adminApiServerConfig.go @@ -17,7 +17,7 @@ func (h *Handlers) GetServerConfig(w http.ResponseWriter, r *http.Request) { ffmpeg := utils.ValidatedFfmpegPath(configRepository.GetFfMpegPath()) usernameBlocklist := configRepository.GetForbiddenUsernameList() usernameSuggestions := configRepository.GetSuggestedUsernamesList() - c := config.GetConfig() + c := config.Get() videoQualityVariants := make([]models.StreamOutputVariant, 0) for _, variant := range configRepository.GetStreamOutputVariants() { diff --git a/webserver/handlers/adminApiSoftwareUpdate.go b/webserver/handlers/adminApiSoftwareUpdate.go index d39c25d6d..ac54c6920 100644 --- a/webserver/handlers/adminApiSoftwareUpdate.go +++ b/webserver/handlers/adminApiSoftwareUpdate.go @@ -44,7 +44,7 @@ func (h *Handlers) AutoUpdateOptions(w http.ResponseWriter, r *http.Request) { CanRestart: false, } - c := config.GetConfig() + c := config.Get() // Nothing is supported when running under "dev" or the feature is // explicitly disabled. diff --git a/webserver/handlers/adminApiSystemStatus.go b/webserver/handlers/adminApiSystemStatus.go index ea2e7798e..a4cb53ce5 100644 --- a/webserver/handlers/adminApiSystemStatus.go +++ b/webserver/handlers/adminApiSystemStatus.go @@ -4,28 +4,30 @@ import ( "encoding/json" "net/http" - "github.com/owncast/owncast/core" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/metrics" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/webserver/middleware" log "github.com/sirupsen/logrus" ) // Status gets the details of the inbound broadcaster. func (h *Handlers) GetAdminStatus(w http.ResponseWriter, r *http.Request) { - broadcaster := core.GetBroadcaster() - status := core.GetStatus() - currentBroadcast := core.GetCurrentBroadcast() - health := metrics.GetStreamHealthOverview() + s := status.Get() + m := metrics.Get() + + broadcaster := s.GetBroadcaster() + currentBroadcast := s.GetCurrentBroadcast() + health := m.GetStreamHealthOverview() response := adminStatusResponse{ Broadcaster: broadcaster, CurrentBroadcast: currentBroadcast, - Online: status.Online, + Online: s.Online, Health: health, - ViewerCount: status.ViewerCount, - OverallPeakViewerCount: status.OverallMaxViewerCount, - SessionPeakViewerCount: status.SessionMaxViewerCount, - VersionNumber: status.VersionNumber, + ViewerCount: s.ViewerCount, + OverallPeakViewerCount: s.OverallMaxViewerCount, + SessionPeakViewerCount: s.SessionMaxViewerCount, + VersionNumber: s.VersionNumber, StreamTitle: configRepository.GetStreamTitle(), } diff --git a/webserver/handlers/adminApiVideoConfig.go b/webserver/handlers/adminApiVideoConfig.go index 3b39d67ff..5d2fd1107 100644 --- a/webserver/handlers/adminApiVideoConfig.go +++ b/webserver/handlers/adminApiVideoConfig.go @@ -4,38 +4,41 @@ import ( "encoding/json" "net/http" - "github.com/owncast/owncast/core" + "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/metrics" + "github.com/owncast/owncast/services/status" log "github.com/sirupsen/logrus" ) // GetVideoPlaybackMetrics returns video playback metrics. func (h *Handlers) GetVideoPlaybackMetrics(w http.ResponseWriter, r *http.Request) { type response struct { - Errors []metrics.TimestampedValue `json:"errors"` - QualityVariantChanges []metrics.TimestampedValue `json:"qualityVariantChanges"` + Errors []models.TimestampedValue `json:"errors"` + QualityVariantChanges []models.TimestampedValue `json:"qualityVariantChanges"` - HighestLatency []metrics.TimestampedValue `json:"highestLatency"` - MedianLatency []metrics.TimestampedValue `json:"medianLatency"` - LowestLatency []metrics.TimestampedValue `json:"lowestLatency"` + HighestLatency []models.TimestampedValue `json:"highestLatency"` + MedianLatency []models.TimestampedValue `json:"medianLatency"` + LowestLatency []models.TimestampedValue `json:"lowestLatency"` - MedianDownloadDuration []metrics.TimestampedValue `json:"medianSegmentDownloadDuration"` - MaximumDownloadDuration []metrics.TimestampedValue `json:"maximumSegmentDownloadDuration"` - MinimumDownloadDuration []metrics.TimestampedValue `json:"minimumSegmentDownloadDuration"` + MedianDownloadDuration []models.TimestampedValue `json:"medianSegmentDownloadDuration"` + MaximumDownloadDuration []models.TimestampedValue `json:"maximumSegmentDownloadDuration"` + MinimumDownloadDuration []models.TimestampedValue `json:"minimumSegmentDownloadDuration"` - SlowestDownloadRate []metrics.TimestampedValue `json:"minPlayerBitrate"` - MedianDownloadRate []metrics.TimestampedValue `json:"medianPlayerBitrate"` - HighestDownloadRater []metrics.TimestampedValue `json:"maxPlayerBitrate"` - AvailableBitrates []int `json:"availableBitrates"` - SegmentLength int `json:"segmentLength"` - Representation int `json:"representation"` + SlowestDownloadRate []models.TimestampedValue `json:"minPlayerBitrate"` + MedianDownloadRate []models.TimestampedValue `json:"medianPlayerBitrate"` + HighestDownloadRater []models.TimestampedValue `json:"maxPlayerBitrate"` + AvailableBitrates []int `json:"availableBitrates"` + SegmentLength int `json:"segmentLength"` + Representation int `json:"representation"` } + s := status.Get() + availableBitrates := []int{} var segmentLength int - if core.GetCurrentBroadcast() != nil { - segmentLength = core.GetCurrentBroadcast().LatencyLevel.SecondsPerSegment - for _, variants := range core.GetCurrentBroadcast().OutputSettings { + if s.GetCurrentBroadcast() != nil { + segmentLength = s.GetCurrentBroadcast().LatencyLevel.SecondsPerSegment + for _, variants := range s.GetCurrentBroadcast().OutputSettings { availableBitrates = append(availableBitrates, variants.VideoBitrate) } } else { @@ -45,21 +48,23 @@ func (h *Handlers) GetVideoPlaybackMetrics(w http.ResponseWriter, r *http.Reques } } - errors := metrics.GetPlaybackErrorCountOverTime() - medianLatency := metrics.GetMedianLatencyOverTime() - minimumLatency := metrics.GetMinimumLatencyOverTime() - maximumLatency := metrics.GetMaximumLatencyOverTime() + m := metrics.Get() + + errors := m.GetPlaybackErrorCountOverTime() + medianLatency := m.GetMedianLatencyOverTime() + minimumLatency := m.GetMinimumLatencyOverTime() + maximumLatency := m.GetMaximumLatencyOverTime() - medianDurations := metrics.GetMedianDownloadDurationsOverTime() - maximumDurations := metrics.GetMaximumDownloadDurationsOverTime() - minimumDurations := metrics.GetMinimumDownloadDurationsOverTime() + medianDurations := m.GetMedianDownloadDurationsOverTime() + maximumDurations := m.GetMaximumDownloadDurationsOverTime() + minimumDurations := m.GetMinimumDownloadDurationsOverTime() - minPlayerBitrate := metrics.GetSlowestDownloadRateOverTime() - medianPlayerBitrate := metrics.GetMedianDownloadRateOverTime() - maxPlayerBitrate := metrics.GetMaxDownloadRateOverTime() - qualityVariantChanges := metrics.GetQualityVariantChangesOverTime() + minPlayerBitrate := m.GetSlowestDownloadRateOverTime() + medianPlayerBitrate := m.GetMedianDownloadRateOverTime() + maxPlayerBitrate := m.GetMaxDownloadRateOverTime() + qualityVariantChanges := m.GetQualityVariantChangesOverTime() - representation := metrics.GetPlaybackMetricsRepresentation() + representation := m.GetPlaybackMetricsRepresentation() resp := response{ AvailableBitrates: availableBitrates, diff --git a/webserver/handlers/adminApiViewers.go b/webserver/handlers/adminApiViewers.go index cbf737d2c..48dd9becf 100644 --- a/webserver/handlers/adminApiViewers.go +++ b/webserver/handlers/adminApiViewers.go @@ -25,7 +25,9 @@ func (h *Handlers) GetViewersOverTime(w http.ResponseWriter, r *http.Request) { windowStartAt := time.Unix(int64(windowStartAtUnix), 0) windowEnd := time.Now() - viewersOverTime := metrics.GetViewersOverTime(windowStartAt, windowEnd) + m := metrics.Get() + + viewersOverTime := m.GetViewersOverTime(windowStartAt, windowEnd) w.Header().Set("Content-Type", "application/json") err = json.NewEncoder(w).Encode(viewersOverTime) if err != nil { diff --git a/webserver/handlers/auth/fediverse/fediverse.go b/webserver/handlers/auth/fediverse/fediverse.go index 0d96dcce9..e863a3b06 100644 --- a/webserver/handlers/auth/fediverse/fediverse.go +++ b/webserver/handlers/auth/fediverse/fediverse.go @@ -5,23 +5,36 @@ import ( "fmt" "net/http" - "github.com/owncast/owncast/activitypub" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/apfederation" fediverseauth "github.com/owncast/owncast/services/auth/fediverse" + "github.com/owncast/owncast/services/chat" + "github.com/owncast/owncast/storage/chatrepository" "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/storage/userrepository" "github.com/owncast/owncast/webserver/responses" log "github.com/sirupsen/logrus" ) -var ( - userRepository = userrepository.Get() - configRepository = configrepository.Get() -) +type FediAuthHandlers struct { + configRepository *configrepository.SqlConfigRepository + chatService *chat.Chat + chatRepository *chatrepository.ChatRepository + userRepository *userrepository.SqlUserRepository +} + +// New creates a new instances of web server handlers. +func New() *FediAuthHandlers { + return &FediAuthHandlers{ + configRepository: configrepository.Get(), + chatService: chat.Get(), + chatRepository: chatrepository.Get(), + userRepository: userrepository.Get(), + } +} // RegisterFediverseOTPRequest registers a new OTP request for the given access token. -func RegisterFediverseOTPRequest(u models.User, w http.ResponseWriter, r *http.Request) { +func (h *FediAuthHandlers) RegisterFediverseOTPRequest(u models.User, w http.ResponseWriter, r *http.Request) { type request struct { FediverseAccount string `json:"account"` } @@ -45,7 +58,9 @@ func RegisterFediverseOTPRequest(u models.User, w http.ResponseWriter, r *http.R return } - msg := fmt.Sprintf("<p>This is an automated message from %s. If you did not request this message please ignore or block. Your requested one-time code is:</p><p>%s</p>", configRepository.GetServerName(), reg.Code) + activitypub := apfederation.Get() + + msg := fmt.Sprintf("<p>This is an automated message from %s. If you did not request this message please ignore or block. Your requested one-time code is:</p><p>%s</p>", h.configRepository.GetServerName(), reg.Code) if err := activitypub.SendDirectFederatedMessage(msg, reg.Account); err != nil { responses.WriteSimpleResponse(w, false, "Could not send code to fediverse: "+err.Error()) return @@ -55,7 +70,7 @@ func RegisterFediverseOTPRequest(u models.User, w http.ResponseWriter, r *http.R } // VerifyFediverseOTPRequest verifies the given OTP code for the given access token. -func VerifyFediverseOTPRequest(w http.ResponseWriter, r *http.Request) { +func (h *FediAuthHandlers) VerifyFediverseOTPRequest(w http.ResponseWriter, r *http.Request) { type request struct { Code string `json:"code"` } @@ -76,20 +91,20 @@ func VerifyFediverseOTPRequest(w http.ResponseWriter, r *http.Request) { } // Check if a user with this auth already exists, if so, log them in. - if u := userRepository.GetUserByAuth(authRegistration.Account, models.Fediverse); u != nil { + if u := h.userRepository.GetUserByAuth(authRegistration.Account, models.Fediverse); u != nil { // Handle existing auth. log.Debugln("user with provided fedvierse identity already exists, logging them in") // Update the current user's access token to point to the existing user id. userID := u.ID - if err := userRepository.SetAccessTokenToOwner(accessToken, userID); err != nil { + if err := h.userRepository.SetAccessTokenToOwner(accessToken, userID); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } if authRegistration.UserDisplayName != u.DisplayName { loginMessage := fmt.Sprintf("**%s** is now authenticated as **%s**", authRegistration.UserDisplayName, u.DisplayName) - if err := chat.SendSystemAction(loginMessage, true); err != nil { + if err := h.chatService.SendSystemAction(loginMessage, true); err != nil { log.Errorln(err) } } @@ -101,14 +116,14 @@ func VerifyFediverseOTPRequest(w http.ResponseWriter, r *http.Request) { // Otherwise, save this as new auth. log.Debug("fediverse account does not already exist, saving it as a new one for the current user") - if err := userRepository.AddAuth(authRegistration.UserID, authRegistration.Account, models.Fediverse); err != nil { + if err := h.userRepository.AddAuth(authRegistration.UserID, authRegistration.Account, models.Fediverse); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } // Update the current user's authenticated flag so we can show it in // the chat UI. - if err := userRepository.SetUserAsAuthenticated(authRegistration.UserID); err != nil { + if err := h.userRepository.SetUserAsAuthenticated(authRegistration.UserID); err != nil { log.Errorln(err) } diff --git a/webserver/handlers/auth/indieauth/client.go b/webserver/handlers/auth/indieauth/client.go index d4199f274..5fa3238cf 100644 --- a/webserver/handlers/auth/indieauth/client.go +++ b/webserver/handlers/auth/indieauth/client.go @@ -6,22 +6,35 @@ import ( "io" "net/http" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" ia "github.com/owncast/owncast/services/auth/indieauth" + "github.com/owncast/owncast/services/chat" + "github.com/owncast/owncast/storage/chatrepository" "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/storage/userrepository" "github.com/owncast/owncast/webserver/responses" log "github.com/sirupsen/logrus" ) -var ( - userRepository = userrepository.Get() - configRepository = configrepository.Get() -) +type IndieAuthHandlers struct { + configRepository *configrepository.SqlConfigRepository + chatService *chat.Chat + chatRepository *chatrepository.ChatRepository + userRepository *userrepository.SqlUserRepository +} + +// New creates a new instances of web server handlers. +func New() *IndieAuthHandlers { + return &IndieAuthHandlers{ + configRepository: configrepository.Get(), + chatService: chat.Get(), + chatRepository: chatrepository.Get(), + userRepository: userrepository.Get(), + } +} // StartAuthFlow will begin the IndieAuth flow for the current user. -func StartAuthFlow(u models.User, w http.ResponseWriter, r *http.Request) { +func (h *IndieAuthHandlers) StartAuthFlow(u models.User, w http.ResponseWriter, r *http.Request) { type request struct { AuthHost string `json:"authHost"` } @@ -59,7 +72,7 @@ func StartAuthFlow(u models.User, w http.ResponseWriter, r *http.Request) { // HandleRedirect will handle the redirect from an IndieAuth server to // continue the auth flow. -func HandleRedirect(w http.ResponseWriter, r *http.Request) { +func (h *IndieAuthHandlers) HandleRedirect(w http.ResponseWriter, r *http.Request) { indieAuthClient := ia.GetIndieAuthClient() state := r.URL.Query().Get("state") code := r.URL.Query().Get("code") @@ -72,21 +85,21 @@ func HandleRedirect(w http.ResponseWriter, r *http.Request) { } // Check if a user with this auth already exists, if so, log them in. - if u := userRepository.GetUserByAuth(response.Me, models.IndieAuth); u != nil { + if u := h.userRepository.GetUserByAuth(response.Me, models.IndieAuth); u != nil { // Handle existing auth. log.Debugln("user with provided indieauth already exists, logging them in") // Update the current user's access token to point to the existing user id. accessToken := request.CurrentAccessToken userID := u.ID - if err := userRepository.SetAccessTokenToOwner(accessToken, userID); err != nil { + if err := h.userRepository.SetAccessTokenToOwner(accessToken, userID); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } if request.DisplayName != u.DisplayName { loginMessage := fmt.Sprintf("**%s** is now authenticated as **%s**", request.DisplayName, u.DisplayName) - if err := chat.SendSystemAction(loginMessage, true); err != nil { + if err := h.chatService.SendSystemAction(loginMessage, true); err != nil { log.Errorln(err) } } @@ -98,14 +111,14 @@ func HandleRedirect(w http.ResponseWriter, r *http.Request) { // Otherwise, save this as new auth. log.Debug("indieauth token does not already exist, saving it as a new one for the current user") - if err := userRepository.AddAuth(request.UserID, response.Me, models.IndieAuth); err != nil { + if err := h.userRepository.AddAuth(request.UserID, response.Me, models.IndieAuth); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return } // Update the current user's authenticated flag so we can show it in // the chat UI. - if err := userRepository.SetUserAsAuthenticated(request.UserID); err != nil { + if err := h.userRepository.SetUserAsAuthenticated(request.UserID); err != nil { log.Errorln(err) } diff --git a/webserver/handlers/auth/indieauth/server.go b/webserver/handlers/auth/indieauth/server.go index 9396bea5a..c361e1e98 100644 --- a/webserver/handlers/auth/indieauth/server.go +++ b/webserver/handlers/auth/indieauth/server.go @@ -10,21 +10,21 @@ import ( ) // HandleAuthEndpoint will handle the IndieAuth auth endpoint. -func HandleAuthEndpoint(w http.ResponseWriter, r *http.Request) { +func (h *IndieAuthHandlers) HandleAuthEndpoint(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet { // Require the GET request for IndieAuth to be behind admin login. - f := middleware.RequireAdminAuth(handleAuthEndpointGet) + f := middleware.RequireAdminAuth(h.handleAuthEndpointGet) f(w, r) return } else if r.Method == http.MethodPost { - handleAuthEndpointPost(w, r) + h.handleAuthEndpointPost(w, r) } else { w.WriteHeader(http.StatusMethodNotAllowed) return } } -func handleAuthEndpointGet(w http.ResponseWriter, r *http.Request) { +func (h *IndieAuthHandlers) handleAuthEndpointGet(w http.ResponseWriter, r *http.Request) { clientID := r.URL.Query().Get("client_id") redirectURI := r.URL.Query().Get("redirect_uri") codeChallenge := r.URL.Query().Get("code_challenge") @@ -58,7 +58,7 @@ func handleAuthEndpointGet(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, u.String(), http.StatusTemporaryRedirect) } -func handleAuthEndpointPost(w http.ResponseWriter, r *http.Request) { +func (h *IndieAuthHandlers) handleAuthEndpointPost(w http.ResponseWriter, r *http.Request) { if err := r.ParseForm(); err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return diff --git a/webserver/handlers/chat.go b/webserver/handlers/chat.go index 45daebae8..7a7ce64dc 100644 --- a/webserver/handlers/chat.go +++ b/webserver/handlers/chat.go @@ -5,7 +5,6 @@ import ( "errors" "net/http" - "github.com/owncast/owncast/core/chat" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" "github.com/owncast/owncast/storage/configrepository" @@ -18,21 +17,21 @@ import ( // ExternalGetChatMessages gets all of the chat messages. func (h *Handlers) ExternalGetChatMessages(integration models.ExternalAPIUser, w http.ResponseWriter, r *http.Request) { middleware.EnableCors(w) - getChatMessages(w, r) + h.getChatMessages(w, r) } // GetChatMessages gets all of the chat messages. func (h *Handlers) GetChatMessages(u models.User, w http.ResponseWriter, r *http.Request) { middleware.EnableCors(w) - getChatMessages(w, r) + h.getChatMessages(w, r) } -func getChatMessages(w http.ResponseWriter, r *http.Request) { +func (h *Handlers) getChatMessages(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.Method { case http.MethodGet: - messages := chat.GetChatHistory() + messages := h.chatRepository.GetChatHistory() if err := json.NewEncoder(w).Encode(messages); err != nil { log.Debugln(err) diff --git a/webserver/handlers/config.go b/webserver/handlers/config.go index c8b44c0b9..432a1fb3d 100644 --- a/webserver/handlers/config.go +++ b/webserver/handlers/config.go @@ -6,9 +6,9 @@ import ( "net/http" "net/url" - "github.com/owncast/owncast/activitypub" "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/federationrepository" "github.com/owncast/owncast/utils" "github.com/owncast/owncast/webserver/middleware" "github.com/owncast/owncast/webserver/responses" @@ -87,7 +87,9 @@ func getConfigResponse() webConfigResponse { var federationResponse federationConfigResponse federationEnabled := configRepository.GetFederationEnabled() - followerCount, _ := activitypub.GetFollowerCount() + federationRepository := federationrepository.Get() + + followerCount, _ := federationRepository.GetFollowerCount() if federationEnabled { serverURLString := configRepository.GetServerURL() serverURL, _ := url.Parse(serverURLString) @@ -117,7 +119,7 @@ func getConfigResponse() webConfigResponse { IndieAuthEnabled: configRepository.GetServerURL() != "", } - c := config.GetConfig() + c := config.Get() return webConfigResponse{ Name: configRepository.GetServerName(), diff --git a/webserver/handlers/disconnect.go b/webserver/handlers/disconnect.go index 898c7fcd7..6c277d55a 100644 --- a/webserver/handlers/disconnect.go +++ b/webserver/handlers/disconnect.go @@ -3,7 +3,7 @@ package handlers import ( "net/http" - "github.com/owncast/owncast/core" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/webserver/responses" "github.com/owncast/owncast/video/rtmp" @@ -11,7 +11,9 @@ import ( // DisconnectInboundConnection will force-disconnect an inbound stream. func (h *Handlers) DisconnectInboundConnection(w http.ResponseWriter, r *http.Request) { - if !core.GetStatus().Online { + s := status.Get() + + if !s.Online { responses.WriteSimpleResponse(w, false, "no inbound stream connected") return } diff --git a/webserver/handlers/emoji.go b/webserver/handlers/emoji.go index 4a6a4f90b..1b50f15ef 100644 --- a/webserver/handlers/emoji.go +++ b/webserver/handlers/emoji.go @@ -26,8 +26,7 @@ func (h *Handlers) GetCustomEmojiList(w http.ResponseWriter, r *http.Request) { func (h *Handlers) GetCustomEmojiImage(w http.ResponseWriter, r *http.Request) { path := strings.TrimPrefix(r.URL.Path, "/img/emoji/") r.URL.Path = path - - c := config.GetConfig() + c := config.Get() emojiFS := os.DirFS(c.CustomEmojiPath) middleware.SetCachingHeaders(w, r) diff --git a/activitypub/controllers/actors.go b/webserver/handlers/federation/actors.go index 8a7109e90..6a245bf0e 100644 --- a/activitypub/controllers/actors.go +++ b/webserver/handlers/federation/actors.go @@ -1,4 +1,4 @@ -package controllers +package federation import ( "net/http" @@ -6,16 +6,17 @@ import ( log "github.com/sirupsen/logrus" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/requests" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" "github.com/owncast/owncast/storage/configrepository" ) -var configRepository = configrepository.Get() - // ActorHandler handles requests for a single actor. func ActorHandler(w http.ResponseWriter, r *http.Request) { + configRepository := configrepository.Get() + req := requests.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -52,7 +53,7 @@ func ActorHandler(w http.ResponseWriter, r *http.Request) { publicKey := crypto.GetPublicKey(actorIRI) person := apmodels.MakeServiceForAccount(accountName) - if err := requests.WriteStreamResponse(person, w, publicKey); err != nil { + if err := req.WriteStreamResponse(person, w, publicKey); err != nil { log.Errorln("unable to write stream response for actor handler", err) w.WriteHeader(http.StatusInternalServerError) return diff --git a/activitypub/controllers/followers.go b/webserver/handlers/federation/followers.go index 0a7a982be..34fb9b7d9 100644 --- a/activitypub/controllers/followers.go +++ b/webserver/handlers/federation/followers.go @@ -1,4 +1,4 @@ -package controllers +package federation import ( "fmt" @@ -7,15 +7,16 @@ import ( "strconv" "strings" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/federationrepository" "github.com/pkg/errors" log "github.com/sirupsen/logrus" "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/requests" ) const ( @@ -53,13 +54,16 @@ func FollowersHandler(w http.ResponseWriter, r *http.Request) { actorIRI := apmodels.MakeLocalIRIForAccount(accountName) publicKey := crypto.GetPublicKey(actorIRI) - if err := requests.WriteStreamResponse(response.(vocab.Type), w, publicKey); err != nil { + req := requests.Get() + if err := req.WriteStreamResponse(response.(vocab.Type), w, publicKey); err != nil { log.Errorln("unable to write stream response for followers handler", err) } } func getInitialFollowersRequest(r *http.Request) (vocab.ActivityStreamsOrderedCollection, error) { - followerCount, _ := persistence.GetFollowerCount() + federationRespository := federationrepository.Get() + + followerCount, _ := federationRespository.GetFollowerCount() collection := streams.NewActivityStreamsOrderedCollection() idProperty := streams.NewJSONLDIdProperty() id, err := createPageURL(r, nil) @@ -92,12 +96,14 @@ func getFollowersPage(page string, r *http.Request) (vocab.ActivityStreamsOrdere return nil, errors.Wrap(err, "unable to parse page number") } - followerCount, err := persistence.GetFollowerCount() + federationRespository := federationrepository.Get() + + followerCount, err := federationRespository.GetFollowerCount() if err != nil { return nil, errors.Wrap(err, "unable to get follower count") } - followers, _, err := persistence.GetFederationFollowers(followersPageSize, (pageInt-1)*followersPageSize) + followers, _, err := federationRespository.GetFederationFollowers(followersPageSize, (pageInt-1)*followersPageSize) if err != nil { return nil, errors.Wrap(err, "unable to get federation followers") } @@ -144,6 +150,8 @@ func getFollowersPage(page string, r *http.Request) (vocab.ActivityStreamsOrdere } func createPageURL(r *http.Request, page *string) (*url.URL, error) { + configRepository := configrepository.Get() + domain := configRepository.GetServerURL() if domain == "" { return nil, errors.New("unable to get server URL") diff --git a/activitypub/controllers/inbox.go b/webserver/handlers/federation/inbox.go index 651c9cd2a..356a3da71 100644 --- a/activitypub/controllers/inbox.go +++ b/webserver/handlers/federation/inbox.go @@ -1,12 +1,13 @@ -package controllers +package federation import ( "io" "net/http" "strings" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/inbox" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/inbox" + "github.com/owncast/owncast/storage/configrepository" log "github.com/sirupsen/logrus" ) @@ -21,6 +22,8 @@ func InboxHandler(w http.ResponseWriter, r *http.Request) { } func acceptInboxRequest(w http.ResponseWriter, r *http.Request) { + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -49,7 +52,8 @@ func acceptInboxRequest(w http.ResponseWriter, r *http.Request) { return } + ib := inbox.Get() inboxRequest := apmodels.InboxRequest{Request: r, ForLocalAccount: forLocalAccount, Body: data} - inbox.AddToQueue(inboxRequest) + ib.AddToQueue(inboxRequest) w.WriteHeader(http.StatusAccepted) } diff --git a/activitypub/controllers/nodeinfo.go b/webserver/handlers/federation/nodeinfo.go index 0953e79f9..da68f18a7 100644 --- a/activitypub/controllers/nodeinfo.go +++ b/webserver/handlers/federation/nodeinfo.go @@ -1,15 +1,16 @@ -package controllers +package federation import ( "fmt" "net/http" "net/url" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/requests" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" "github.com/owncast/owncast/services/config" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/federationrepository" log "github.com/sirupsen/logrus" ) @@ -24,6 +25,8 @@ func NodeInfoController(w http.ResponseWriter, r *http.Request) { Links []links `json:"links"` } + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -88,13 +91,16 @@ func NodeInfoV2Controller(w http.ResponseWriter, r *http.Request) { Metadata metadata `json:"metadata"` } + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return } - localPostCount, _ := persistence.GetLocalPostCount() - c := config.GetConfig() + federationRespository := federationrepository.Get() + localPostCount, _ := federationRespository.GetLocalPostCount() + c := config.Get() res := response{ Version: "2.0", @@ -163,6 +169,8 @@ func XNodeInfo2Controller(w http.ResponseWriter, r *http.Request) { OpenRegistrations bool `json:"openRegistrations"` } + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -174,8 +182,9 @@ func XNodeInfo2Controller(w http.ResponseWriter, r *http.Request) { return } - localPostCount, _ := persistence.GetLocalPostCount() - c := config.GetConfig() + federationRespository := federationrepository.Get() + localPostCount, _ := federationRespository.GetLocalPostCount() + c := config.Get() res := &response{ Organization: Organization{ @@ -233,6 +242,8 @@ func InstanceV1Controller(w http.ResponseWriter, r *http.Request) { InvitesEnabled bool `json:"invites_enabled"` } + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -251,8 +262,9 @@ func InstanceV1Controller(w http.ResponseWriter, r *http.Request) { } thumbnail.Path = "/logo/external" - localPostCount, _ := persistence.GetLocalPostCount() - c := config.GetConfig() + federationRespository := federationrepository.Get() + localPostCount, _ := federationRespository.GetLocalPostCount() + c := config.Get() res := response{ URI: serverURL, @@ -277,15 +289,20 @@ func InstanceV1Controller(w http.ResponseWriter, r *http.Request) { } func writeResponse(payload interface{}, w http.ResponseWriter) error { + configRepository := configrepository.Get() + accountName := configRepository.GetDefaultFederationUsername() actorIRI := apmodels.MakeLocalIRIForAccount(accountName) publicKey := crypto.GetPublicKey(actorIRI) - return requests.WritePayloadResponse(payload, w, publicKey) + req := requests.Get() + return req.WritePayloadResponse(payload, w, publicKey) } // HostMetaController points to webfinger. func HostMetaController(w http.ResponseWriter, r *http.Request) { + configRepository := configrepository.Get() + serverURL := configRepository.GetServerURL() if serverURL == "" { w.WriteHeader(http.StatusNotFound) diff --git a/activitypub/controllers/object.go b/webserver/handlers/federation/object.go index 82a963e8c..e9ee81e96 100644 --- a/activitypub/controllers/object.go +++ b/webserver/handlers/federation/object.go @@ -1,18 +1,21 @@ -package controllers +package federation import ( "net/http" "strings" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/requests" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" + "github.com/owncast/owncast/storage/configrepository" + "github.com/owncast/owncast/storage/federationrepository" log "github.com/sirupsen/logrus" ) // ObjectHandler handles requests for a single federated ActivityPub object. func ObjectHandler(w http.ResponseWriter, r *http.Request) { + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -24,8 +27,9 @@ func ObjectHandler(w http.ResponseWriter, r *http.Request) { return } + federationRespository := federationrepository.Get() iri := strings.Join([]string{strings.TrimSuffix(configRepository.GetServerURL(), "/"), r.URL.Path}, "") - object, _, _, err := persistence.GetObjectByIRI(iri) + object, _, _, err := federationRespository.GetObjectByIRI(iri) if err != nil { w.WriteHeader(http.StatusNotFound) return @@ -35,7 +39,8 @@ func ObjectHandler(w http.ResponseWriter, r *http.Request) { actorIRI := apmodels.MakeLocalIRIForAccount(accountName) publicKey := crypto.GetPublicKey(actorIRI) - if err := requests.WriteResponse([]byte(object), w, publicKey); err != nil { + req := requests.Get() + if err := req.WriteResponse([]byte(object), w, publicKey); err != nil { log.Errorln(err) } } diff --git a/activitypub/controllers/outbox.go b/webserver/handlers/federation/outbox.go index 0dbcb5a03..1eee5da0a 100644 --- a/activitypub/controllers/outbox.go +++ b/webserver/handlers/federation/outbox.go @@ -1,4 +1,4 @@ -package controllers +package federation import ( "fmt" @@ -8,10 +8,11 @@ import ( "github.com/go-fed/activity/streams" "github.com/go-fed/activity/streams/vocab" - "github.com/owncast/owncast/activitypub/apmodels" - "github.com/owncast/owncast/activitypub/crypto" - "github.com/owncast/owncast/activitypub/persistence" - "github.com/owncast/owncast/activitypub/requests" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/services/apfederation/crypto" + "github.com/owncast/owncast/services/apfederation/requests" + "github.com/owncast/owncast/storage/federationrepository" + "github.com/pkg/errors" log "github.com/sirupsen/logrus" ) @@ -51,14 +52,17 @@ func OutboxHandler(w http.ResponseWriter, r *http.Request) { actorIRI := apmodels.MakeLocalIRIForAccount(accountName) publicKey := crypto.GetPublicKey(actorIRI) - if err := requests.WriteStreamResponse(response.(vocab.Type), w, publicKey); err != nil { + req := requests.Get() + if err := req.WriteStreamResponse(response.(vocab.Type), w, publicKey); err != nil { log.Errorln("unable to write stream response for outbox handler", err) } } // ActorObjectHandler will handle the request for a single ActivityPub object. func ActorObjectHandler(w http.ResponseWriter, r *http.Request) { - object, _, _, err := persistence.GetObjectByIRI(r.URL.Path) + federationRespository := federationrepository.Get() + + object, _, _, err := federationRespository.GetObjectByIRI(r.URL.Path) if err != nil { w.WriteHeader(http.StatusNotFound) return @@ -81,7 +85,9 @@ func getInitialOutboxHandler(r *http.Request) (vocab.ActivityStreamsOrderedColle idProperty.SetIRI(id) collection.SetJSONLDId(idProperty) - totalPosts, err := persistence.GetOutboxPostCount() + federationRespository := federationrepository.Get() + + totalPosts, err := federationRespository.GetOutboxPostCount() if err != nil { return nil, errors.Wrap(err, "unable to get outbox post count") } @@ -108,7 +114,9 @@ func getOutboxPage(page string, r *http.Request) (vocab.ActivityStreamsOrderedCo return nil, errors.Wrap(err, "unable to parse page number") } - postCount, err := persistence.GetOutboxPostCount() + federationRespository := federationrepository.Get() + + postCount, err := federationRespository.GetOutboxPostCount() if err != nil { return nil, errors.Wrap(err, "unable to get outbox post count") } @@ -124,7 +132,7 @@ func getOutboxPage(page string, r *http.Request) (vocab.ActivityStreamsOrderedCo orderedItems := streams.NewActivityStreamsOrderedItemsProperty() - outboxItems, err := persistence.GetOutbox(outboxPageSize, (pageInt-1)*outboxPageSize) + outboxItems, err := federationRespository.GetOutbox(outboxPageSize, (pageInt-1)*outboxPageSize) if err != nil { return nil, errors.Wrap(err, "unable to get federation followers") } diff --git a/activitypub/controllers/webfinger.go b/webserver/handlers/federation/webfinger.go index 11b347ebe..740eeb037 100644 --- a/activitypub/controllers/webfinger.go +++ b/webserver/handlers/federation/webfinger.go @@ -1,17 +1,20 @@ -package controllers +package federation import ( "encoding/json" "net/http" "strings" - "github.com/owncast/owncast/activitypub/apmodels" + "github.com/owncast/owncast/services/apfederation/apmodels" + "github.com/owncast/owncast/storage/configrepository" "github.com/owncast/owncast/utils" log "github.com/sirupsen/logrus" ) // WebfingerHandler will handle webfinger lookup requests. func WebfingerHandler(w http.ResponseWriter, r *http.Request) { + configRepository := configrepository.Get() + if !configRepository.GetFederationEnabled() { w.WriteHeader(http.StatusMethodNotAllowed) log.Debugln("webfinger request rejected! Federation is not enabled") diff --git a/webserver/handlers/followers.go b/webserver/handlers/followers.go index 61b502676..e64a1582c 100644 --- a/webserver/handlers/followers.go +++ b/webserver/handlers/followers.go @@ -3,13 +3,14 @@ package handlers import ( "net/http" - "github.com/owncast/owncast/activitypub/persistence" + "github.com/owncast/owncast/storage/federationrepository" "github.com/owncast/owncast/webserver/responses" ) // GetFollowers will handle an API request to fetch the list of followers (non-activitypub response). func (h *Handlers) GetFollowers(offset int, limit int, w http.ResponseWriter, r *http.Request) { - followers, total, err := persistence.GetFederationFollowers(limit, offset) + federationRepository := federationrepository.Get() + followers, total, err := federationRepository.GetFederationFollowers(limit, offset) if err != nil { responses.WriteSimpleResponse(w, false, "unable to fetch followers") return diff --git a/webserver/handlers/handlers.go b/webserver/handlers/handlers.go index 0d23597a3..4e2db1062 100644 --- a/webserver/handlers/handlers.go +++ b/webserver/handlers/handlers.go @@ -1,8 +1,22 @@ package handlers -type Handlers struct{} +import ( + "github.com/owncast/owncast/services/chat" + "github.com/owncast/owncast/storage/chatrepository" + "github.com/owncast/owncast/storage/configrepository" +) + +type Handlers struct { + configRepository *configrepository.SqlConfigRepository + chatService *chat.Chat + chatRepository *chatrepository.ChatRepository +} // New creates a new instances of web server handlers. func New() *Handlers { - return &Handlers{} + return &Handlers{ + configRepository: configrepository.Get(), + chatService: chat.Get(), + chatRepository: chatrepository.Get(), + } } diff --git a/webserver/handlers/hls.go b/webserver/handlers/hls.go index 4744d9045..0bbaa58a5 100644 --- a/webserver/handlers/hls.go +++ b/webserver/handlers/hls.go @@ -22,7 +22,7 @@ func (h *Handlers) HandleHLSRequest(w http.ResponseWriter, r *http.Request) { return } - c := config.GetConfig() + c := config.Get() requestedPath := r.URL.Path relativePath := strings.Replace(requestedPath, "/hls/", "", 1) diff --git a/webserver/handlers/images.go b/webserver/handlers/images.go index 8d2171fb3..5e7b765b7 100644 --- a/webserver/handlers/images.go +++ b/webserver/handlers/images.go @@ -16,7 +16,7 @@ const ( // GetThumbnail will return the thumbnail image as a response. func (h *Handlers) GetThumbnail(w http.ResponseWriter, r *http.Request) { - c := config.GetConfig() + c := config.Get() imageFilename := "thumbnail.jpg" imagePath := filepath.Join(c.TempDir, imageFilename) @@ -41,7 +41,7 @@ func (h *Handlers) GetThumbnail(w http.ResponseWriter, r *http.Request) { // GetPreview will return the preview gif as a response. func (h *Handlers) GetPreview(w http.ResponseWriter, r *http.Request) { - c := config.GetConfig() + c := config.Get() imageFilename := "preview.gif" imagePath := filepath.Join(c.TempDir, imageFilename) diff --git a/webserver/handlers/moderation.go b/webserver/handlers/moderation.go index 6d1419002..152ccf1d2 100644 --- a/webserver/handlers/moderation.go +++ b/webserver/handlers/moderation.go @@ -6,8 +6,6 @@ import ( "strings" "time" - "github.com/owncast/owncast/core/chat" - "github.com/owncast/owncast/core/chat/events" "github.com/owncast/owncast/models" "github.com/owncast/owncast/webserver/responses" log "github.com/sirupsen/logrus" @@ -26,7 +24,7 @@ func (h *Handlers) GetUserDetails(w http.ResponseWriter, r *http.Request) { type response struct { User *models.User `json:"user"` ConnectedClients []connectedClient `json:"connectedClients"` - Messages []events.UserMessageEvent `json:"messages"` + Messages []models.UserMessageEvent `json:"messages"` } pathComponents := strings.Split(r.URL.Path, "/") @@ -39,7 +37,7 @@ func (h *Handlers) GetUserDetails(w http.ResponseWriter, r *http.Request) { return } - c, _ := chat.GetClientsForUser(uid) + c, _ := h.chatService.GetClientsForUser(uid) clients := make([]connectedClient, len(c)) for i, c := range c { client := connectedClient{ @@ -55,7 +53,7 @@ func (h *Handlers) GetUserDetails(w http.ResponseWriter, r *http.Request) { clients[i] = client } - messages, err := chat.GetMessagesFromUser(uid) + messages, err := h.chatRepository.GetMessagesFromUser(uid) if err != nil { log.Errorln(err) } diff --git a/webserver/handlers/notifications.go b/webserver/handlers/notifications.go index ebeb8136b..042d6fa49 100644 --- a/webserver/handlers/notifications.go +++ b/webserver/handlers/notifications.go @@ -6,6 +6,7 @@ import ( "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/notifications" + "github.com/owncast/owncast/storage/notificationsrepository" "github.com/owncast/owncast/webserver/responses" "github.com/owncast/owncast/utils" @@ -44,7 +45,9 @@ func (h *Handlers) RegisterForLiveNotifications(u models.User, w http.ResponseWr return } - if err := notifications.AddNotification(req.Channel, req.Destination); err != nil { + n := notificationsrepository.Get() + + if err := n.AddNotification(req.Channel, req.Destination); err != nil { log.Errorln(err) responses.WriteSimpleResponse(w, false, "unable to save notification") return diff --git a/webserver/handlers/playbackMetrics.go b/webserver/handlers/playbackMetrics.go index 4bbfff45e..7ae1de07d 100644 --- a/webserver/handlers/playbackMetrics.go +++ b/webserver/handlers/playbackMetrics.go @@ -35,19 +35,20 @@ func (h *Handlers) ReportPlaybackMetrics(w http.ResponseWriter, r *http.Request) } clientID := utils.GenerateClientIDFromRequest(r) + m := metrics.Get() - metrics.RegisterPlaybackErrorCount(clientID, request.Errors) + m.RegisterPlaybackErrorCount(clientID, request.Errors) if request.Bandwidth != 0.0 { - metrics.RegisterPlayerBandwidth(clientID, request.Bandwidth) + m.RegisterPlayerBandwidth(clientID, request.Bandwidth) } if request.Latency != 0.0 { - metrics.RegisterPlayerLatency(clientID, request.Latency) + m.RegisterPlayerLatency(clientID, request.Latency) } if request.DownloadDuration != 0.0 { - metrics.RegisterPlayerSegmentDownloadDuration(clientID, request.DownloadDuration) + m.RegisterPlayerSegmentDownloadDuration(clientID, request.DownloadDuration) } - metrics.RegisterQualityVariantChangesCount(clientID, request.QualityVariantChanges) + m.RegisterQualityVariantChangesCount(clientID, request.QualityVariantChanges) } diff --git a/webserver/handlers/remoteFollow.go b/webserver/handlers/remoteFollow.go index 12a70d630..3ec021934 100644 --- a/webserver/handlers/remoteFollow.go +++ b/webserver/handlers/remoteFollow.go @@ -7,7 +7,7 @@ import ( "net/url" "strings" - "github.com/owncast/owncast/activitypub/webfinger" + "github.com/owncast/owncast/services/apfederation/webfinger" "github.com/owncast/owncast/webserver/responses" ) @@ -33,10 +33,12 @@ func (h *Handlers) RemoteFollow(w http.ResponseWriter, r *http.Request) { return } + wf := webfinger.Get() + localActorPath, _ := url.Parse(configRepository.GetServerURL()) localActorPath.Path = fmt.Sprintf("/federation/user/%s", configRepository.GetDefaultFederationUsername()) var template string - links, err := webfinger.GetWebfingerLinks(request.Account) + links, err := wf.GetWebfingerLinks(request.Account) if err != nil { responses.WriteSimpleResponse(w, false, err.Error()) return diff --git a/webserver/handlers/status.go b/webserver/handlers/status.go index c6e778b6f..2e7701059 100644 --- a/webserver/handlers/status.go +++ b/webserver/handlers/status.go @@ -5,7 +5,7 @@ import ( "net/http" "time" - "github.com/owncast/owncast/core" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/utils" "github.com/owncast/owncast/webserver/middleware" "github.com/owncast/owncast/webserver/responses" @@ -24,17 +24,17 @@ func (h *Handlers) GetStatus(w http.ResponseWriter, r *http.Request) { } func getStatusResponse() webStatusResponse { - status := core.GetStatus() + s := status.Get() response := webStatusResponse{ - Online: status.Online, + Online: s.Online, ServerTime: time.Now(), - LastConnectTime: status.LastConnectTime, - LastDisconnectTime: status.LastDisconnectTime, - VersionNumber: status.VersionNumber, - StreamTitle: status.StreamTitle, + LastConnectTime: s.Status.LastConnectTime, + LastDisconnectTime: s.Status.LastDisconnectTime, + VersionNumber: s.VersionNumber, + StreamTitle: s.StreamTitle, } if !configRepository.GetHideViewerCount() { - response.ViewerCount = status.ViewerCount + response.ViewerCount = s.ViewerCount } return response } diff --git a/webserver/handlers/ypApi.go b/webserver/handlers/ypApi.go index 910b5039d..cb6c76de5 100644 --- a/webserver/handlers/ypApi.go +++ b/webserver/handlers/ypApi.go @@ -4,8 +4,8 @@ import ( "encoding/json" "net/http" - "github.com/owncast/owncast/core" "github.com/owncast/owncast/models" + "github.com/owncast/owncast/services/status" "github.com/owncast/owncast/utils" log "github.com/sirupsen/logrus" ) @@ -33,7 +33,7 @@ func (h *Handlers) GetYPResponse(w http.ResponseWriter, r *http.Request) { return } - status := core.GetStatus() + s := status.Get() streamTitle := configRepository.GetStreamTitle() @@ -44,11 +44,11 @@ func (h *Handlers) GetYPResponse(w http.ResponseWriter, r *http.Request) { Logo: "/logo", NSFW: configRepository.GetNSFW(), Tags: configRepository.GetServerMetadataTags(), - Online: status.Online, - ViewerCount: status.ViewerCount, - OverallMaxViewerCount: status.OverallMaxViewerCount, - SessionMaxViewerCount: status.SessionMaxViewerCount, - LastConnectTime: status.LastConnectTime, + Online: s.Online, + ViewerCount: s.ViewerCount, + OverallMaxViewerCount: s.OverallMaxViewerCount, + SessionMaxViewerCount: s.SessionMaxViewerCount, + LastConnectTime: s.Status.LastConnectTime, Social: configRepository.GetSocialHandles(), } diff --git a/webserver/middleware/auth.go b/webserver/middleware/auth.go index cb1f4f67f..03370852f 100644 --- a/webserver/middleware/auth.go +++ b/webserver/middleware/auth.go @@ -105,7 +105,7 @@ func RequireExternalAPIAccessToken(scope string, handler ExternalAccessTokenHand // Not to be used for validating 3rd party access. func RequireUserAccessToken(handler UserAccessTokenHandlerFunc) http.HandlerFunc { userRepository := userrepository.Get() - chatRepository := chatrepository.GetChatRepository() + chatRepository := chatrepository.Get() return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { accessToken := r.URL.Query().Get("accessToken") diff --git a/webserver/router.go b/webserver/router.go index 37e788075..f93528a57 100644 --- a/webserver/router.go +++ b/webserver/router.go @@ -3,13 +3,10 @@ package webserver import ( "net/http" - "github.com/owncast/owncast/activitypub" - "github.com/owncast/owncast/core/chat" + "github.com/owncast/owncast/models" "github.com/owncast/owncast/services/config" - "github.com/owncast/owncast/storage" "github.com/owncast/owncast/utils" - fediverseauth "github.com/owncast/owncast/webserver/handlers/auth/fediverse" - "github.com/owncast/owncast/webserver/handlers/auth/indieauth" + "github.com/owncast/owncast/webserver/handlers/federation" "github.com/owncast/owncast/webserver/middleware" "github.com/prometheus/client_golang/prometheus/promhttp" ) @@ -20,12 +17,13 @@ func (s *webServer) setupRoutes() { s.setupAdminAPIRoutes() s.setupExternalThirdPartyAPIRoutes() s.setupModerationAPIRoutes() + s.setupActivityPubFederationRoutes() s.router.HandleFunc("/hls/", s.handlers.HandleHLSRequest) // websocket s.router.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) { - chat.HandleClientConnection(w, r) + s.chatService.HandleClientConnection(w, r) }) s.router.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { @@ -42,13 +40,10 @@ func (s *webServer) setupRoutes() { // s.ServeHTTP(w, r) } }) - - // ActivityPub has its own router - activitypub.Start(data.GetDatastore(), s.router) } func (s *webServer) setupWebAssetRoutes() { - c := config.GetConfig() + c := config.Get() // The admin web app. s.router.HandleFunc("/admin/", middleware.RequireAdminAuth(s.handlers.IndexHandler)) @@ -119,12 +114,12 @@ func (s *webServer) setupInternalAPIRoutes() { s.router.HandleFunc("/api/notifications/register", middleware.RequireUserAccessToken(s.handlers.RegisterForLiveNotifications)) // Start auth flow - s.router.HandleFunc("/api/auth/indieauth", middleware.RequireUserAccessToken(indieauth.StartAuthFlow)) - s.router.HandleFunc("/api/auth/indieauth/callback", indieauth.HandleRedirect) - s.router.HandleFunc("/api/auth/provider/indieauth", indieauth.HandleAuthEndpoint) + s.router.HandleFunc("/api/auth/indieauth", middleware.RequireUserAccessToken(s.indieAuthHandlers.StartAuthFlow)) + s.router.HandleFunc("/api/auth/indieauth/callback", s.indieAuthHandlers.HandleRedirect) + s.router.HandleFunc("/api/auth/provider/indieauth", s.indieAuthHandlers.HandleAuthEndpoint) - s.router.HandleFunc("/api/auth/fediverse", middleware.RequireUserAccessToken(fediverseauth.RegisterFediverseOTPRequest)) - s.router.HandleFunc("/api/auth/fediverse/verify", fediverseauth.VerifyFediverseOTPRequest) + s.router.HandleFunc("/api/auth/fediverse", middleware.RequireUserAccessToken(s.fediAuthHandlers.RegisterFediverseOTPRequest)) + s.router.HandleFunc("/api/auth/fediverse/verify", s.fediAuthHandlers.VerifyFediverseOTPRequest) } func (s *webServer) setupAdminAPIRoutes() { @@ -376,31 +371,31 @@ func (s *webServer) setupAdminAPIRoutes() { func (s *webServer) setupExternalThirdPartyAPIRoutes() { // Send a system message to chat - s.router.HandleFunc("/api/integrations/chat/system", middleware.RequireExternalAPIAccessToken(user.ScopeCanSendSystemMessages, s.handlers.SendSystemMessage)) + s.router.HandleFunc("/api/integrations/chat/system", middleware.RequireExternalAPIAccessToken(models.ScopeCanSendSystemMessages, s.handlers.SendSystemMessage)) // Send a system message to a single client - s.router.HandleFunc(utils.RestEndpoint("/api/integrations/chat/system/client/{clientId}", middleware.RequireExternalAPIAccessToken(user.ScopeCanSendSystemMessages, s.handlers.SendSystemMessageToConnectedClient))) + s.router.HandleFunc(utils.RestEndpoint("/api/integrations/chat/system/client/{clientId}", middleware.RequireExternalAPIAccessToken(models.ScopeCanSendSystemMessages, s.handlers.SendSystemMessageToConnectedClient))) // Send a user message to chat *NO LONGER SUPPORTED - s.router.HandleFunc("/api/integrations/chat/user", middleware.RequireExternalAPIAccessToken(user.ScopeCanSendChatMessages, s.handlers.SendUserMessage)) + s.router.HandleFunc("/api/integrations/chat/user", middleware.RequireExternalAPIAccessToken(models.ScopeCanSendChatMessages, s.handlers.SendUserMessage)) // Send a message to chat as a specific 3rd party bot/integration based on its access token - s.router.HandleFunc("/api/integrations/chat/send", middleware.RequireExternalAPIAccessToken(user.ScopeCanSendChatMessages, s.handlers.SendIntegrationChatMessage)) + s.router.HandleFunc("/api/integrations/chat/send", middleware.RequireExternalAPIAccessToken(models.ScopeCanSendChatMessages, s.handlers.SendIntegrationChatMessage)) // Send a user action to chat - s.router.HandleFunc("/api/integrations/chat/action", middleware.RequireExternalAPIAccessToken(user.ScopeCanSendSystemMessages, s.handlers.SendChatAction)) + s.router.HandleFunc("/api/integrations/chat/action", middleware.RequireExternalAPIAccessToken(models.ScopeCanSendSystemMessages, s.handlers.SendChatAction)) // Hide chat message - s.router.HandleFunc("/api/integrations/chat/messagevisibility", middleware.RequireExternalAPIAccessToken(user.ScopeHasAdminAccess, s.handlers.ExternalUpdateMessageVisibility)) + s.router.HandleFunc("/api/integrations/chat/messagevisibility", middleware.RequireExternalAPIAccessToken(models.ScopeHasAdminAccess, s.handlers.ExternalUpdateMessageVisibility)) // Stream title - s.router.HandleFunc("/api/integrations/streamtitle", middleware.RequireExternalAPIAccessToken(user.ScopeHasAdminAccess, s.handlers.ExternalSetStreamTitle)) + s.router.HandleFunc("/api/integrations/streamtitle", middleware.RequireExternalAPIAccessToken(models.ScopeHasAdminAccess, s.handlers.ExternalSetStreamTitle)) // Get chat history - s.router.HandleFunc("/api/integrations/chat", middleware.RequireExternalAPIAccessToken(user.ScopeHasAdminAccess, s.handlers.ExternalGetChatMessages)) + s.router.HandleFunc("/api/integrations/chat", middleware.RequireExternalAPIAccessToken(models.ScopeHasAdminAccess, s.handlers.ExternalGetChatMessages)) // Connected clients - s.router.HandleFunc("/api/integrations/clients", middleware.RequireExternalAPIAccessToken(user.ScopeHasAdminAccess, s.handlers.ExternalGetConnectedChatClients)) + s.router.HandleFunc("/api/integrations/clients", middleware.RequireExternalAPIAccessToken(models.ScopeHasAdminAccess, s.handlers.ExternalGetConnectedChatClients)) } func (s *webServer) setupModerationAPIRoutes() { @@ -413,3 +408,30 @@ func (s *webServer) setupModerationAPIRoutes() { // Get a user's details s.router.HandleFunc("/api/moderation/chat/user/", middleware.RequireUserModerationScopeAccesstoken(s.handlers.GetUserDetails)) } + +// StartRouter will start the federation specific http router. +func (s *webServer) setupActivityPubFederationRoutes() { + // WebFinger + s.router.HandleFunc("/.well-known/webfinger", federation.WebfingerHandler) + + // Host Metadata + s.router.HandleFunc("/.well-known/host-meta", federation.HostMetaController) + + // Nodeinfo v1 + s.router.HandleFunc("/.well-known/nodeinfo", federation.NodeInfoController) + + // x-nodeinfo v2 + s.router.HandleFunc("/.well-known/x-nodeinfo2", federation.XNodeInfo2Controller) + + // Nodeinfo v2 + s.router.HandleFunc("/nodeinfo/2.0", federation.NodeInfoV2Controller) + + // Instance details + s.router.HandleFunc("/api/v1/instance", federation.InstanceV1Controller) + + // Single ActivityPub Actor + s.router.HandleFunc("/federation/user/", middleware.RequireActivityPubOrRedirect(federation.ActorHandler)) + + // Single AP object + s.router.HandleFunc("/federation/", middleware.RequireActivityPubOrRedirect(federation.ObjectHandler)) +} diff --git a/webserver/webserver.go b/webserver/webserver.go index d9f433344..526a9a89e 100644 --- a/webserver/webserver.go +++ b/webserver/webserver.go @@ -6,21 +6,32 @@ import ( "time" "github.com/CAFxX/httpcompression" + "github.com/owncast/owncast/services/chat" "github.com/owncast/owncast/webserver/handlers" + "github.com/owncast/owncast/webserver/handlers/auth/fediverse" + "github.com/owncast/owncast/webserver/handlers/auth/indieauth" + log "github.com/sirupsen/logrus" "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" ) type webServer struct { - router *http.ServeMux - handlers *handlers.Handlers - server *http.Server + router *http.ServeMux + handlers *handlers.Handlers + fediAuthHandlers *fediverse.FediAuthHandlers + indieAuthHandlers *indieauth.IndieAuthHandlers + chatService *chat.Chat + + server *http.Server } func New() *webServer { s := &webServer{ - router: http.NewServeMux(), + router: http.NewServeMux(), + handlers: handlers.New(), + fediAuthHandlers: fediverse.New(), + indieAuthHandlers: indieauth.New(), } s.setupRoutes() diff --git a/webserver/webserver_test.go b/webserver/webserver_test.go index 92bd1bc5d..4386757a4 100644 --- a/webserver/webserver_test.go +++ b/webserver/webserver_test.go @@ -3,19 +3,18 @@ package webserver import ( "net/http" "net/http/httptest" - "os" "testing" + + "github.com/owncast/owncast/storage/data" ) var srv *webServer func TestMain(m *testing.M) { - dbFile, err := os.CreateTemp(os.TempDir(), "owncast-test-db.db") + _, err := data.NewStore(":memory:") if err != nil { panic(err) } - - data.SetupPersistence(dbFile.Name()) srv = New() m.Run() |