Merge pull request #394 from sbondCo/jellyfin-sync

Jellyfin (Manual) Sync

server/activity.go

@@ -11,23 +11,27 @@ import (
 type ActivityType string
 
 var (
-	ADDED_WATCHED          ActivityType = "ADDED_WATCHED"
-	REMOVED_WATCHED        ActivityType = "REMOVED_WATCHED"
-	RATING_CHANGED         ActivityType = "RATING_CHANGED"
-	STATUS_CHANGED         ActivityType = "STATUS_CHANGED"
-	THOUGHTS_CHANGED       ActivityType = "THOUGHTS_CHANGED"
-	THOUGHTS_REMOVED       ActivityType = "THOUGHTS_REMOVED"
-	IMPORTED_WATCHED       ActivityType = "IMPORTED_WATCHED"
-	IMPORTED_RATING        ActivityType = "IMPORTED_RATING"        // Imported rating, but with no rating acts as original import of content to old platform (where they are importing from) activity
-	IMPORTED_ADDED_WATCHED ActivityType = "IMPORTED_ADDED_WATCHED" // Imported watched date, so we can save the original watch dates of content from users old platform (where they are importing from).
-	SEASON_ADDED           ActivityType = "SEASON_ADDED"
-	SEASON_REMOVED         ActivityType = "SEASON_REMOVED"
-	SEASON_RATING_CHANGED  ActivityType = "SEASON_RATING_CHANGED"
-	SEASON_STATUS_CHANGED  ActivityType = "SEASON_STATUS_CHANGED"
-	EPISODE_ADDED          ActivityType = "EPISODE_ADDED"
-	EPISODE_REMOVED        ActivityType = "EPISODE_REMOVED"
-	EPISODE_RATING_CHANGED ActivityType = "EPISODE_RATING_CHANGED"
-	EPISODE_STATUS_CHANGED ActivityType = "EPISODE_STATUS_CHANGED"
+	ADDED_WATCHED             ActivityType = "ADDED_WATCHED"
+	REMOVED_WATCHED           ActivityType = "REMOVED_WATCHED"
+	RATING_CHANGED            ActivityType = "RATING_CHANGED"
+	STATUS_CHANGED            ActivityType = "STATUS_CHANGED"
+	THOUGHTS_CHANGED          ActivityType = "THOUGHTS_CHANGED"
+	THOUGHTS_REMOVED          ActivityType = "THOUGHTS_REMOVED"
+	IMPORTED_WATCHED          ActivityType = "IMPORTED_WATCHED"
+	IMPORTED_WATCHED_JF       ActivityType = "IMPORTED_WATCHED_JF"
+	IMPORTED_RATING           ActivityType = "IMPORTED_RATING"        // Imported rating, but with no rating acts as original import of content to old platform (where they are importing from) activity
+	IMPORTED_ADDED_WATCHED    ActivityType = "IMPORTED_ADDED_WATCHED" // Imported watched date, so we can save the original watch dates of content from users old platform (where they are importing from).
+	IMPORTED_ADDED_WATCHED_JF ActivityType = "IMPORTED_ADDED_WATCHED_JF"
+	SEASON_ADDED              ActivityType = "SEASON_ADDED"
+	SEASON_ADDED_JF           ActivityType = "SEASON_ADDED_JF"
+	SEASON_REMOVED            ActivityType = "SEASON_REMOVED"
+	SEASON_RATING_CHANGED     ActivityType = "SEASON_RATING_CHANGED"
+	SEASON_STATUS_CHANGED     ActivityType = "SEASON_STATUS_CHANGED"
+	EPISODE_ADDED             ActivityType = "EPISODE_ADDED"
+	EPISODE_ADDED_JF          ActivityType = "EPISODE_ADDED_JF"
+	EPISODE_REMOVED           ActivityType = "EPISODE_REMOVED"
+	EPISODE_RATING_CHANGED    ActivityType = "EPISODE_RATING_CHANGED"
+	EPISODE_STATUS_CHANGED    ActivityType = "EPISODE_STATUS_CHANGED"
 )
 
 type Activity struct {

server/jellyfin.go

@@ -8,6 +8,9 @@ import (
 	"log/slog"
 	"net/http"
 	"net/url"
+	"time"
+
+	"github.com/gin-gonic/gin"
 )
 
 type JellyfinItemSearchResponse struct {
@@ -15,12 +18,27 @@ type JellyfinItemSearchResponse struct {
 }
 
 type JellyfinItems struct {
+	Name        string `json:"Name"`
 	Type        string `json:"Type"`
 	ServerID    string `json:"ServerId"`
 	Id          string `json:"Id"`
 	ProviderIds struct {
 		Tmdb string `json:"Tmdb"`
 	} `json:"ProviderIds"`
+	UserData struct {
+		Rating                float64   `json:"Rating"`
+		PlayedPercentage      float64   `json:"PlayedPercentage"`
+		UnplayedItemCount     int64     `json:"UnplayedItemCount"`
+		PlaybackPositionTicks int64     `json:"PlaybackPositionTicks"`
+		PlayCount             int64     `json:"PlayCount"`
+		IsFavorite            bool      `json:"IsFavorite"`
+		Likes                 bool      `json:"Likes"`
+		LastPlayedDate        time.Time `json:"LastPlayedDate"`
+		Played                bool      `json:"Played"`
+		Key                   string    `json:"Key"`
+		ItemId                string    `json:"ItemId"`
+	} `json:"UserData"`
+	RecursiveItemCount int64 `json:"RecursiveItemCount"`
 }
 
 type JFContentFindResponse struct {
@@ -28,6 +46,33 @@ type JFContentFindResponse struct {
 	Url        string `json:"url"`
 }
 
+// Jellyfin access middleware, ensures user is a jellyfin user.
+// To be ran after AuthRequired middleware with extra data.
+func JellyfinAccessRequired() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		userId := c.MustGet("userId").(uint)
+		slog.Debug("JellyfinAccessRequired middleware hit", "user_id", userId)
+		userType := c.MustGet("userType").(UserType)
+		userThirdPartyId := c.MustGet("userThirdPartyId").(string)
+		userThirdPartyAuth := c.MustGet("userThirdPartyAuth").(string)
+		if Config.JELLYFIN_HOST == "" {
+			slog.Error("JellyfinAccessRequired: Request made to login via Jellyfin, but JELLYFIN_HOST has not been configured.")
+			c.AbortWithStatus(401)
+			return
+		}
+		if userType != JELLYFIN_USER || userThirdPartyId == "" {
+			slog.Error("JellyfinAccessRequired: User is not a jellyfin user..", "user_type", userType, "user_third_party_id", userThirdPartyId)
+			c.AbortWithStatus(401)
+			return
+		}
+		if userThirdPartyAuth == "" {
+			slog.Error("JellyfinAccessRequired: User has no thirdPartyAuth token..")
+			c.AbortWithStatus(401)
+			return
+		}
+	}
+}
+
 func jellyfinAPIRequest(method string, ep string, p map[string]string, username string, userToken string, resp interface{}) error {
 	if Config.JELLYFIN_HOST == "" {
 		slog.Error("jellyfinAPIRequest: JELLYFIN_HOST not configured.")
@@ -97,18 +142,6 @@ func jellyfinContentFind(
 	contentName string,
 	contentTmdbId string,
 ) (JFContentFindResponse, error) {
-	if Config.JELLYFIN_HOST == "" {
-		slog.Error("Request made to login via Jellyfin, but JELLYFIN_HOST has not been configured.")
-		return JFContentFindResponse{}, errors.New("jellyfin login not enabled")
-	}
-	if userType != JELLYFIN_USER || userThirdPartyId == "" {
-		slog.Error("User is not a jellyfin user..", "user_type", userType, "user_third_party_id", userThirdPartyId)
-		return JFContentFindResponse{}, errors.New("not jellyfin user")
-	}
-	if userThirdPartyAuth == "" {
-		slog.Error("User has no thirdPartyAuth token..")
-		return JFContentFindResponse{}, errors.New("user has no jellyfin auth token")
-	}
 	if contentType == "" || contentName == "" {
 		slog.Error("Bad request", "content_type", contentType, "content_name", contentName)
 		return JFContentFindResponse{}, errors.New("content type or name not provided")

server/jellyfin_sync.go

@@ -0,0 +1,307 @@
+package main
+
+import (
+	"errors"
+	"log/slog"
+	"strconv"
+
+	"gorm.io/gorm"
+)
+
+type JellyfinSeriesSeasonsResponse struct {
+	Items []JellyfinSeriesSeasonItem `json:"Items"`
+}
+
+type JellyfinSeriesSeasonItem struct {
+	JellyfinItems
+	// aka the season number
+	IndexNumber int `json:"IndexNumber"`
+}
+
+type JellyfinSeriesEpisodesResponse struct {
+	Items []JellyfinSeriesEpisodeItem `json:"Items"`
+}
+
+type JellyfinSeriesEpisodeItem struct {
+	JellyfinItems
+	// the episode number
+	IndexNumber int `json:"IndexNumber"`
+	// the episodes season number
+	ParentIndexNumber int `json:"ParentIndexNumber"`
+}
+
+type JellyfinSyncResponse struct {
+	JobId string `json:"jobId"`
+}
+
+// Perform the jellyfin sync.
+// Gets each type of media separately from jellyfin and attempts to import them.
+// Errors are added silently to the job.
+func startJellyfinSync(
+	db *gorm.DB,
+	jobId string,
+	userId uint,
+	username string,
+	userThirdPartyId string,
+	userThirdPartyAuth string,
+) {
+	// Get played movies
+	updateJobCurrentTask(jobId, userId, "syncing movies")
+	playedMovies := new(JellyfinItemSearchResponse)
+	err := jellyfinAPIRequest(
+		"GET",
+		"/Users/"+userThirdPartyId+"/Items",
+		map[string]string{
+			"Filters":          "IsPlayed",
+			"IncludeItemTypes": "Movie",
+			"Fields":           "ProviderIds",
+			"Recursive":        "true",
+		},
+		username,
+		userThirdPartyAuth,
+		&playedMovies,
+	)
+	if err != nil {
+		slog.Error("jellyfinSyncWatched: Jellyfin API request failed", "error", err)
+		addJobError(jobId, userId, "failed to get jellyfin response for movies")
+	} else {
+		if len(playedMovies.Items) <= 0 {
+			slog.Info("jellyfinSyncWatched: User has no played movies.", "user_id", userId)
+		} else {
+			for _, v := range playedMovies.Items {
+				slog.Info("jellyfinSyncWatched: Importing played movie.", "movie_name", v.Name, "user_id", userId)
+				slog.Debug("jellyfinSyncWatched: Importing played movie.", "full_item", v, "user_id", userId)
+
+				// 1. Ensure we have a tmdbId
+				if v.ProviderIds.Tmdb == "" {
+					slog.Error("jellyfinSyncWatched: Movie to import does not have a tmdb id.", "movie_name", v.Name, "movie_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "movie could not be imported (no tmdbId present): "+v.Name)
+					continue
+				}
+				tmdbId, err := strconv.Atoi(v.ProviderIds.Tmdb)
+				if err != nil {
+					slog.Error("jellyfinSyncWatched: Movie to import does not have a parseable (to int) tmdb id.", "movie_name", v.Name, "movie_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "movie could not be imported (tmdbId was not parseable): "+v.Name)
+					continue
+				}
+
+				updateJobCurrentTask(jobId, userId, "syncing "+v.Name)
+
+				// 2. Imported watched movie
+				w, err := addWatched(db, userId, WatchedAddRequest{
+					Status:      FINISHED,
+					ContentID:   tmdbId,
+					ContentType: MOVIE,
+					WatchedDate: v.UserData.LastPlayedDate,
+				}, IMPORTED_WATCHED_JF)
+				if err != nil {
+					if err.Error() == "content already on watched list" {
+						slog.Error("jellyfinSyncWatched: Unique constraint hit.. content must already be on watch list.", "movie_name", v.Name, "movie_ids", v.ProviderIds, "user_id", userId)
+					} else {
+						slog.Error("jellyfinSyncWatched: Movie failed to import.", "movie_name", v.Name, "movie_ids", v.ProviderIds, "user_id", userId)
+						addJobError(jobId, userId, "movie could not be imported (failed when adding to watched list): "+v.Name)
+					}
+				} else {
+					// 3. Add IMPORTED_ADDED_WATCHED_JF activity
+					if !v.UserData.LastPlayedDate.IsZero() {
+						_, err := addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: IMPORTED_ADDED_WATCHED_JF, CustomDate: &v.UserData.LastPlayedDate})
+						if err != nil {
+							slog.Error("jellyfinSyncWatched: Failed to add dateswatched activity.", "movie_name", v.Name,
+								"movie_ids", v.ProviderIds, "user_id", userId, "date", v.UserData.LastPlayedDate, "error", err)
+						}
+					}
+				}
+			}
+		}
+	}
+
+	// Get played series
+	// Can't rely on IsPlayed filter, since we want to get partially played series too.
+	updateJobCurrentTask(jobId, userId, "syncing series")
+	allSeries := new(JellyfinItemSearchResponse)
+	err = jellyfinAPIRequest(
+		"GET",
+		"/Users/"+userThirdPartyId+"/Items",
+		map[string]string{
+			"IncludeItemTypes": "Series",
+			"Fields":           "ProviderIds,RecursiveItemCount",
+			"Recursive":        "true",
+			"IsPlaceHolder":    "false",
+		},
+		username,
+		userThirdPartyAuth,
+		&allSeries,
+	)
+	if err != nil {
+		slog.Error("jellyfinSyncWatched: Jellyfin API request failed", "error", err)
+		addJobError(jobId, userId, "failed to get jellyfin response for series")
+	} else {
+		if len(allSeries.Items) <= 0 {
+			slog.Info("jellyfinSyncWatched: No series found.", "user_id", userId)
+		} else {
+			// Import series
+			for _, v := range allSeries.Items {
+				slog.Info("jellyfinSyncWatched: Processing series.", "series_name", v.Name, "user_id", userId)
+				slog.Debug("jellyfinSyncWatched: Processing series.", "full_item", v, "user_id", userId)
+
+				// 1. Make sure show is watched or at least partially watched
+				if !v.UserData.Played && v.UserData.PlayedPercentage <= 0 && v.RecursiveItemCount == v.UserData.UnplayedItemCount {
+					slog.Debug("jellyfinSyncWatched: Skipping unwatched series:", "series_name", v.Name, "user_id", userId)
+					continue
+				}
+
+				// 1.1. Ensure we have a tmdbId
+				if v.ProviderIds.Tmdb == "" {
+					slog.Error("jellyfinSyncWatched: Series to import does not have a tmdb id.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "series could not be imported (no tmdbId present): "+v.Name)
+					continue
+				}
+				tmdbId, err := strconv.Atoi(v.ProviderIds.Tmdb)
+				if err != nil {
+					slog.Error("jellyfinSyncWatched: Series to import does not have a parseable (to int) tmdb id.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "series could not be imported (tmdbId was not parseable): "+v.Name)
+					continue
+				}
+
+				updateJobCurrentTask(jobId, userId, "syncing serie "+v.Name)
+
+				// 2. Imported watched series
+				w, err := addWatched(db, userId, WatchedAddRequest{
+					Status:      FINISHED,
+					ContentID:   tmdbId,
+					ContentType: SHOW,
+					WatchedDate: v.UserData.LastPlayedDate,
+				}, IMPORTED_WATCHED_JF)
+				if err != nil {
+					if err.Error() == "content already on watched list" {
+						slog.Info("jellyfinSyncWatched: Unique constraint hit.. content must already be on watch list.",
+							"series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId, "watched_id", w.ID)
+					} else {
+						slog.Error("jellyfinSyncWatched: Series failed to import.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+						addJobError(jobId, userId, "series could not be imported (failed when adding to watched list): "+v.Name)
+					}
+				} else {
+					// 3. Add IMPORTED_ADDED_WATCHED activity (only if no err above, show also must not have already been on our list)
+					if !v.UserData.LastPlayedDate.IsZero() {
+						_, err := addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: IMPORTED_ADDED_WATCHED_JF, CustomDate: &v.UserData.LastPlayedDate})
+						if err != nil {
+							slog.Error("jellyfinSyncWatched: Failed to add dateswatched activity.", "series_name", v.Name,
+								"series_ids", v.ProviderIds, "user_id", userId, "date", v.UserData.LastPlayedDate, "error", err)
+						}
+					}
+				}
+
+				// 4. Import watched seasons for this serie
+				// Get all show seasons (filtering isPlayed doesn't seem to be a thing, so we will have to do that ourselves)
+				seriesSeasons := new(JellyfinSeriesSeasonsResponse)
+				err = jellyfinAPIRequest(
+					"GET",
+					"/Shows/"+v.Id+"/Seasons",
+					map[string]string{
+						"UserId":        userThirdPartyId,
+						"Fields":        "ProviderIds",
+						"IsPlaceHolder": "false",
+					},
+					username,
+					userThirdPartyAuth,
+					&seriesSeasons,
+				)
+				if err != nil {
+					slog.Error("jellyfinSyncWatched: Failed to fetch series seasons.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "series seasons could not be imported (request failed): "+v.Name)
+				} else if len(seriesSeasons.Items) <= 0 {
+					slog.Info("jellyfinSyncWatched: Series has no seasons.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+				} else {
+					for _, vs := range seriesSeasons.Items {
+						if !vs.UserData.Played {
+							slog.Debug("jellyfinSyncWatched: Skipping import of unplayed season.", "series_name", v.Name, "season_num", vs.IndexNumber, "user_id", userId)
+							continue
+						}
+						updateJobCurrentTask(jobId, userId, "syncing "+v.Name+" season "+strconv.Itoa(vs.IndexNumber))
+						_, err = addWatchedSeason(db, userId, WatchedSeasonAddRequest{
+							WatchedID:    w.ID,
+							SeasonNumber: vs.IndexNumber,
+							Status:       FINISHED,
+							addActivity:  SEASON_ADDED_JF,
+						})
+						if err != nil {
+							slog.Error("jellyfinSyncWatched: Failed to fetch series seasons.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+							addJobError(jobId, userId, "series season could not be imported (addWatchedSeason request failed): "+v.Name+" season "+strconv.Itoa(vs.IndexNumber))
+						}
+					}
+				}
+
+				// 5. Import watched episodes for this serie
+				// Gets all show episodes (filtering isPlayed doesn't seem to be a thing, so we will have to do that ourselves)
+				seriesEpisodes := new(JellyfinSeriesEpisodesResponse)
+				err = jellyfinAPIRequest(
+					"GET",
+					"/Shows/"+v.Id+"/Episodes",
+					map[string]string{
+						"UserId":        userThirdPartyId,
+						"Fields":        "ProviderIds",
+						"IsPlaceHolder": "false",
+					},
+					username,
+					userThirdPartyAuth,
+					&seriesEpisodes,
+				)
+				if err != nil {
+					slog.Error("jellyfinSyncWatched: Failed to fetch series episodes.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+					addJobError(jobId, userId, "series episodes could not be imported (request failed): "+v.Name)
+				} else if len(seriesEpisodes.Items) <= 0 {
+					slog.Info("jellyfinSyncWatched: Series has no episodes.", "series_name", v.Name, "series_ids", v.ProviderIds, "user_id", userId)
+				} else {
+					for _, vs := range seriesEpisodes.Items {
+						if !vs.UserData.Played {
+							slog.Debug("jellyfinSyncWatched: Skipping import of unplayed episode.", "series_name", v.Name, "season_num", vs.ParentIndexNumber, "episode_num", vs.IndexNumber, "user_id", userId)
+							continue
+						}
+						updateJobCurrentTask(jobId, userId, "syncing "+v.Name+" season "+strconv.Itoa(vs.ParentIndexNumber)+" episode "+strconv.Itoa(vs.IndexNumber))
+						_, err = addWatchedEpisodes(db, userId, WatchedEpisodeAddRequest{
+							WatchedID:     w.ID,
+							SeasonNumber:  vs.ParentIndexNumber,
+							EpisodeNumber: vs.IndexNumber,
+							Status:        FINISHED,
+							addActivity:   EPISODE_ADDED_JF,
+						})
+						if err != nil {
+							slog.Error("jellyfinSyncWatched: Failed to import series episode.", "series_name", v.Name, "season_num", vs.ParentIndexNumber, "episode_num", vs.IndexNumber, "user_id", userId)
+							addJobError(jobId, userId, "series episode could not be imported (addWatchedEpisode request failed): "+v.Name+" "+vs.Name)
+						}
+					}
+				}
+			}
+		}
+	}
+
+	updateJobStatus(jobId, userId, JOB_DONE)
+}
+
+func jellyfinSyncWatched(
+	db *gorm.DB,
+	userId uint,
+	userType UserType,
+	username string,
+	userThirdPartyId string,
+	userThirdPartyAuth string,
+) (JellyfinSyncResponse, error) {
+	jobId, err := addJob("jf_sync", userId)
+	if err != nil {
+		slog.Error("jellyfinSyncWatched: Failed to create a job", "error", err)
+		return JellyfinSyncResponse{}, errors.New("failed to create job")
+	}
+
+	updateJobStatus(jobId, userId, JOB_RUNNING)
+
+	go startJellyfinSync(
+		db,
+		jobId,
+		userId,
+		username,
+		userThirdPartyId,
+		userThirdPartyAuth,
+	)
+
+	return JellyfinSyncResponse{JobId: jobId}, nil
+}

server/jobs.go

@@ -0,0 +1,125 @@
+// We will use jobs only for storing/retrieving active job statuses for the client.
+// Running the job will be done wherever needed, but isn't handled here.
+// When starting a job elsewhere, we should first add a job here as active to get an `id`,
+// this id should be used to update the active job so the client can request job status updates.
+
+package main
+
+import (
+	"errors"
+	"log/slog"
+	"time"
+)
+
+type JobStatus string
+
+var (
+	JOB_CREATED   JobStatus = "CREATED"
+	JOB_RUNNING   JobStatus = "RUNNING"
+	JOB_DONE      JobStatus = "DONE"
+	JOB_CANCELLED JobStatus = "CANCELLED"
+)
+
+type Job struct {
+	// We can give the job a name simply for showing on the client.
+	Name string `json:"name"`
+	// The current status of the job.
+	Status JobStatus `json:"status"`
+	// The current task we are performing inside the job.
+	// Just so we can portray progress on the client by displaying the current task.
+	CurrentTask string `json:"currentTask,omitempty"`
+	// Errors that occurred in the task
+	Errors []string `json:"errors"`
+	// Stored for access control.
+	UserId uint `json:"-"`
+}
+
+var activeJobs = make(map[string]*Job)
+
+// Add a job to our activeJobs map.
+// Returns id of job on success, or error if failed to add.
+func addJob(name string, userId uint) (string, error) {
+	idk, err := generateString(8)
+	if err != nil {
+		return "", err
+	}
+	_, ok := activeJobs[idk]
+	if ok {
+		// Lets just hope this doesn't happen, may the odds be with us.
+		return "", errors.New("job already exists with id generated, please try again")
+	}
+	activeJobs[idk] = &Job{
+		Name:   name,
+		Status: JOB_CREATED,
+		UserId: userId,
+	}
+	return idk, nil
+}
+
+func rmJob(id string, userId uint) {
+	slog.Debug("rmJob: Removing a job.", "id", id)
+	v, ok := activeJobs[id]
+	if ok && v.UserId == userId {
+		delete(activeJobs, id)
+		slog.Debug("rmJob: Removed a job.", "id", id)
+		return
+	}
+	slog.Debug("rmJob: Job to remove does not exist (or not owned by this user).", "id", id, "user_id", userId)
+}
+
+// Get a job.
+// Returns job if found, otherwise errors if job does not exist.
+func getJob(id string, userId uint) (*Job, error) {
+	j, ok := activeJobs[id]
+	if ok {
+		// Ensure user requesting a job, owns the job.
+		if j.UserId != userId {
+			slog.Warn("getJob: A user tried to access a job they do not own.", "user_id", userId, "job_id", id)
+			return &Job{}, errors.New("job does not exist")
+		}
+		return j, nil
+	}
+	return &Job{}, errors.New("job does not exist")
+}
+
+// Update a jobs status.
+func updateJobStatus(id string, userId uint, status JobStatus) error {
+	j, err := getJob(id, userId)
+	if err != nil {
+		slog.Error("updateJobStatus: Failed!", "status", status, "error", err)
+		return err
+	}
+	j.Status = status
+	// If job is set to done, remove it after 1 minute.
+	if status == JOB_DONE {
+		slog.Debug("updateJobStatus: Job set to done. Will be removed after 1m.", "id", id)
+		go func() {
+			time.Sleep(1 * time.Minute)
+			slog.Debug("updateJobStatus: Job done. waited 1m.. removing job now.", "id", id)
+			rmJob(id, userId)
+		}()
+	}
+	return nil
+}
+
+// Update a jobs current task.
+func updateJobCurrentTask(id string, userId uint, ct string) error {
+	j, err := getJob(id, userId)
+	if err != nil {
+		slog.Error("updateJobCurrentTask: Failed!", "ct", ct, "error", err)
+		return err
+	}
+	j.CurrentTask = ct
+	return nil
+}
+
+// Add an error to a job.
+func addJobError(id string, userId uint, e string) error {
+	j, err := getJob(id, userId)
+	if err != nil {
+		slog.Error("updateJobCurrentTask: Failed!", "e", e, "error", err)
+		return err
+	}
+	j.Errors = append(j.Errors, e)
+	return nil
+}

server/routes.go

@@ -663,7 +663,7 @@ func (b *BaseRouter) addProfileRoutes() {
 }
 
 func (b *BaseRouter) addJellyfinRoutes() {
-	jf := b.rg.Group("/jellyfin").Use(AuthRequired(b.db))
+	jf := b.rg.Group("/jellyfin").Use(AuthRequired(b.db), JellyfinAccessRequired())
 
 	// Check if jf has item
 	jf.GET("/:type/:name/:tmdbId", func(c *gin.Context) {
@@ -679,6 +679,21 @@ func (b *BaseRouter) addJellyfinRoutes() {
 		}
 		c.JSON(http.StatusOK, response)
 	})
+
+	// Sync users jellyfin watched items to watchlist
+	jf.GET("/sync", func(c *gin.Context) {
+		userId := c.MustGet("userId").(uint)
+		userType := c.MustGet("userType").(UserType)
+		username := c.MustGet("username").(string)
+		userThirdPartyId := c.MustGet("userThirdPartyId").(string)
+		userThirdPartyAuth := c.MustGet("userThirdPartyAuth").(string)
+		response, err := jellyfinSyncWatched(b.db, userId, userType, username, userThirdPartyId, userThirdPartyAuth)
+		if err != nil {
+			c.JSON(http.StatusForbidden, ErrorResponse{Error: err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, response)
+	})
 }
 
 func (b *BaseRouter) addUserRoutes() {
@@ -1109,3 +1124,17 @@ func (b *BaseRouter) addRadarrRoutes() {
 		c.AbortWithStatusJSON(http.StatusBadRequest, ErrorResponse{Error: err.Error()})
 	})
 }
+
+func (b *BaseRouter) addJobRoutes() {
+	job := b.rg.Group("/job").Use(AuthRequired(nil))
+
+	job.GET("/:id", func(c *gin.Context) {
+		userId := c.MustGet("userId").(uint)
+		response, err := getJob(c.Param("id"), userId)
+		if err != nil {
+			c.JSON(http.StatusForbidden, ErrorResponse{Error: err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, *response)
+	})
+}

server/watcharr.go

@@ -138,6 +138,7 @@ func main() {
 	br.addFeatureRoutes()
 	br.addSonarrRoutes()
 	br.addRadarrRoutes()
+	br.addJobRoutes()
 	br.rg.Static("/img", path.Join(DataPath, "img"))
 
 	go setupTasks(db)

server/watched.go

@@ -163,6 +163,7 @@ func addWatched(db *gorm.DB, userId uint, ar WatchedAddRequest, at ActivityType)
 	}
 	// If custom WatchedDate passed, set CreatedAt and UpdatedAt fields to it.
 	if !ar.WatchedDate.IsZero() {
+		slog.Debug("Adding watched item: The provided WatchedDate is valid.", "watched_date", ar.WatchedDate, "userId", userId, "contentType", ar.ContentType, "contentId", ar.ContentID)
 		watched.CreatedAt = ar.WatchedDate
 		watched.UpdatedAt = ar.WatchedDate
 	}
@@ -174,7 +175,7 @@ func addWatched(db *gorm.DB, userId uint, ar WatchedAddRequest, at ActivityType)
 				return Watched{}, errors.New("content already on watched list. errored checking for soft deleted record")
 			}
 			if watched.DeletedAt.Time.IsZero() {
-				return Watched{}, errors.New("content already on watched list")
+				return watched, errors.New("content already on watched list")
 			} else {
 				slog.Info("addWatched: Watched list item for this content exists as soft deleted record.. attempting to restore")
 				res = db.Model(&Watched{}).Unscoped().Where("user_id = ? AND content_id = ?", userId, watched.ContentID).Updates(map[string]interface{}{"status": ar.Status, "rating": ar.Rating, "deleted_at": nil})

server/watched_episodes.go

@@ -31,6 +31,7 @@ type WatchedEpisodeAddRequest struct {
 	EpisodeNumber int           `json:"episodeNumber"`
 	Status        WatchedStatus `json:"status"`
 	Rating        int8          `json:"rating"`
+	addActivity   ActivityType  `json:"-"`
 }
 
 type WatchedEpisodeAddResponse struct {
@@ -54,16 +55,19 @@ func addWatchedEpisodes(db *gorm.DB, userId uint, ar WatchedEpisodeAddRequest) (
 	if w.Content.Type != SHOW {
 		return WatchedEpisodeAddResponse{}, errors.New("can't add watched episode for non show content")
 	}
-	var found bool
+	found := false
+	updated := false
 	for i, we := range w.WatchedEpisodes {
 		if we.SeasonNumber == ar.SeasonNumber && we.EpisodeNumber == ar.EpisodeNumber {
 			slog.Debug("Existing watched episode item found, updating existing")
 			found = true
-			if ar.Status != "" {
+			if ar.Status != "" && ar.Status != w.WatchedEpisodes[i].Status {
 				w.WatchedEpisodes[i].Status = ar.Status
+				updated = true
 			}
-			if ar.Rating != 0 {
+			if ar.Rating != 0 && ar.Rating != w.WatchedEpisodes[i].Rating {
 				w.WatchedEpisodes[i].Rating = ar.Rating
+				updated = true
 			}
 			break
 		}
@@ -86,17 +90,25 @@ func addWatchedEpisodes(db *gorm.DB, userId uint, ar WatchedEpisodeAddRequest) (
 	}
 	// Add activity
 	if found {
-		if ar.Status != "" {
-			json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "episode": ar.EpisodeNumber, "status": ar.Status})
-			addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: EPISODE_STATUS_CHANGED, Data: string(json)})
-		}
-		if ar.Rating != 0 {
-			json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "episode": ar.EpisodeNumber, "rating": ar.Rating})
-			addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: EPISODE_RATING_CHANGED, Data: string(json)})
+		// Only add change activity if we actually updated a value
+		// (changing value to same value doesn't count).
+		if updated {
+			if ar.Status != "" {
+				json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "episode": ar.EpisodeNumber, "status": ar.Status})
+				addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: EPISODE_STATUS_CHANGED, Data: string(json)})
+			}
+			if ar.Rating != 0 {
+				json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "episode": ar.EpisodeNumber, "rating": ar.Rating})
+				addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: EPISODE_RATING_CHANGED, Data: string(json)})
+			}
 		}
 	} else {
+		at := EPISODE_ADDED
+		if ar.addActivity != "" {
+			at = ar.addActivity
+		}
 		json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "episode": ar.EpisodeNumber, "status": ar.Status, "rating": ar.Rating})
-		addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: EPISODE_ADDED, Data: string(json)})
+		addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: at, Data: string(json)})
 	}
 	return WatchedEpisodeAddResponse{
 		WatchedEpisodes: w.WatchedEpisodes,

server/watched_seasons.go

@@ -25,6 +25,7 @@ type WatchedSeasonAddRequest struct {
 	SeasonNumber int           `json:"seasonNumber"`
 	Status       WatchedStatus `json:"status"`
 	Rating       int8          `json:"rating"`
+	addActivity  ActivityType  `json:"-"`
 }
 
 type WatchedSeasonAddResponse struct {
@@ -48,16 +49,19 @@ func addWatchedSeason(db *gorm.DB, userId uint, ar WatchedSeasonAddRequest) (Wat
 	if w.Content.Type != SHOW {
 		return WatchedSeasonAddResponse{}, errors.New("can't add watched season for non show content")
 	}
-	var found bool
+	found := false
+	updated := false
 	for i, ws := range w.WatchedSeasons {
 		if ws.SeasonNumber == ar.SeasonNumber {
 			slog.Debug("Existing watched season item found, updating existing")
 			found = true
-			if ar.Status != "" {
+			if ar.Status != "" && ar.Status != w.WatchedSeasons[i].Status {
 				w.WatchedSeasons[i].Status = ar.Status
+				updated = true
 			}
-			if ar.Rating != 0 {
+			if ar.Rating != 0 && ar.Rating != w.WatchedSeasons[i].Rating {
 				w.WatchedSeasons[i].Rating = ar.Rating
+				updated = true
 			}
 			break
 		}
@@ -79,17 +83,25 @@ func addWatchedSeason(db *gorm.DB, userId uint, ar WatchedSeasonAddRequest) (Wat
 	}
 	// Add activity
 	if found {
-		if ar.Status != "" {
-			json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "status": ar.Status})
-			addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: SEASON_STATUS_CHANGED, Data: string(json)})
-		}
-		if ar.Rating != 0 {
-			json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "rating": ar.Rating})
-			addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: SEASON_RATING_CHANGED, Data: string(json)})
+		// Only add change activity if we actually updated a value
+		// (changing value to same value doesn't count).
+		if updated {
+			if ar.Status != "" {
+				json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "status": ar.Status})
+				addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: SEASON_STATUS_CHANGED, Data: string(json)})
+			}
+			if ar.Rating != 0 {
+				json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "rating": ar.Rating})
+				addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: SEASON_RATING_CHANGED, Data: string(json)})
+			}
 		}
 	} else {
+		at := SEASON_ADDED
+		if ar.addActivity != "" {
+			at = ar.addActivity
+		}
 		json, _ := json.Marshal(map[string]interface{}{"season": ar.SeasonNumber, "status": ar.Status, "rating": ar.Rating})
-		addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: SEASON_ADDED, Data: string(json)})
+		addedActivity, _ = addActivity(db, userId, ActivityAddRequest{WatchedID: w.ID, Type: at, Data: string(json)})
 	}
 	return WatchedSeasonAddResponse{
 		WatchedSeasons: w.WatchedSeasons,

src/lib/Activity.svelte

@@ -41,6 +41,7 @@
       case "THOUGHTS_REMOVED":
         return "Thoughts Removed";
       case "IMPORTED_WATCHED":
+      case "IMPORTED_WATCHED_JF":
         return "Imported";
       case "IMPORTED_RATING":
         if (a.data) {
@@ -53,8 +54,10 @@
         }
         return "Imported Rating";
       case "IMPORTED_ADDED_WATCHED":
+      case "IMPORTED_ADDED_WATCHED_JF":
         return "Imported Watch Date";
       case "SEASON_ADDED":
+      case "SEASON_ADDED_JF":
         if (a.data) {
           const data = JSON.parse(a.data);
           return `Season ${data.season} Added as ${toFullTitleCase(data.status)}`;
@@ -79,6 +82,7 @@
         }
         return "Season Removed";
       case "EPISODE_ADDED":
+      case "EPISODE_ADDED_JF":
         if (a.data) {
           const data = JSON.parse(a.data);
           return `${seasonAndEpToReadable(data.season, data.episode)} Added ${data.status ? `as ${toFullTitleCase(data.status)}` : data.rating ? `with Rating ${data.rating}` : ""}`;

src/routes/(app)/profile/+page.svelte

@@ -9,11 +9,12 @@
   import { updateUserSetting } from "@/lib/util/api";
   import { getOrdinalSuffix, monthsShort, toggleTheme } from "@/lib/util/helpers";
   import { appTheme, userInfo, userSettings } from "@/store";
-  import type { Image, Profile } from "@/types";
+  import { UserType, type Image, type Profile } from "@/types";
   import axios from "axios";
   import { notify } from "@/lib/util/notify";
   import UserAvatar from "@/lib/img/UserAvatar.svelte";
   import PwChangeModal from "@/routes/(app)/profile/modals/PwChangeModal.svelte";
+  import JellyfinSyncModal from "./modals/JellyfinSyncModal.svelte";
 
   $: user = $userInfo;
   $: settings = $userSettings;
@@ -26,6 +27,7 @@
   let includePreviouslyWatchedDisabled = false;
   let pwChangeModalOpen = false;
   let getProfilePromise = getProfile();
+  let jellyfinSyncModalOpen = false;
 
   async function getProfile() {
     return (await axios.get(`/profile`)).data as Profile;
@@ -261,11 +263,18 @@
       <div class="row btns">
         <button on:click={() => goto("/import")}>Import</button>
         <button on:click={() => downloadWatchedList()} disabled={exportDisabled}>Export</button>
-        <button
-          on:click={() => {
-            pwChangeModalOpen = true;
-          }}>Change Password</button
-        >
+        {#if user?.type !== UserType?.Jellyfin}
+          <button
+            on:click={() => {
+              pwChangeModalOpen = true;
+            }}>Change Password</button
+          >
+        {/if}
+        {#if user?.type === UserType?.Jellyfin}
+          <button on:click={() => (jellyfinSyncModalOpen = true)} disabled={exportDisabled}>
+            Sync With Jellyfin
+          </button>
+        {/if}
       </div>
       {#if pwChangeModalOpen}
         <PwChangeModal
@@ -275,6 +284,9 @@
           }}
         ></PwChangeModal>
       {/if}
+      {#if jellyfinSyncModalOpen}
+        <JellyfinSyncModal onClose={() => (jellyfinSyncModalOpen = false)} />
+      {/if}
     </div>
   </div>
 </div>

src/routes/(app)/profile/modals/JellyfinSyncModal.svelte

@@ -0,0 +1,193 @@
+<script lang="ts">
+  import Icon from "@/lib/Icon.svelte";
+  import Modal from "@/lib/Modal.svelte";
+  import Spinner from "@/lib/Spinner.svelte";
+  import { notify } from "@/lib/util/notify";
+  import { JobStatus, type GetJobResponse, type JellyfinSyncResponse } from "@/types";
+  import axios from "axios";
+  import { onDestroy, onMount } from "svelte";
+  import { watchedList } from "@/store";
+
+  export let onClose: () => void;
+
+  let step: "starting" | "errored" | "job-running" | "done" | "modal-closing" = "starting";
+  let jobId: string | undefined;
+  let currentTask: string | undefined;
+  let latestJobStatus: GetJobResponse | undefined;
+
+  async function startJellyfinSync() {
+    try {
+      const r = await axios.get<JellyfinSyncResponse>("/jellyfin/sync");
+      console.log("startJellyfinSync: Response:", r.data);
+      if (!r.data.jobId) {
+        step = "errored";
+        console.error("startJellyfinSync: No jobId returned!");
+        return;
+      }
+      jobId = r.data.jobId;
+      step = "job-running";
+      startJobWatcher();
+    } catch (err) {
+      console.error("startJellyfinSync failed!", err);
+      step = "errored";
+    }
+  }
+
+  async function startJobWatcher() {
+    if (!jobId) {
+      console.error("startJobWatcher: No Job Id");
+      notify({ text: "Unable to start job watcher, no job id.", type: "error" });
+      return;
+    }
+    console.log("startJobWatcher: Starting..");
+    let seqfailedJobReqs = 0;
+    while (step === "job-running") {
+      try {
+        const r = await axios.get<GetJobResponse>(`/job/${jobId}`);
+        console.log("jobWatcher: Got job data:", r.data);
+        latestJobStatus = r.data;
+        currentTask = r.data?.currentTask;
+        if (r.data?.status === JobStatus.DONE) {
+          step = "done";
+        } else if (r.data?.status === JobStatus.CANCELLED) {
+          step = "errored";
+        }
+        // If we get here without erroring, we can reset it to 0.
+        seqfailedJobReqs = 0;
+      } catch (err) {
+        console.error("jobWatcher: Get job request failed!", seqfailedJobReqs, err);
+        seqfailedJobReqs++;
+      }
+      if (seqfailedJobReqs >= 10) {
+        console.error("jobWatcher: Failed 10 times in a row!");
+        notify({
+          text: "Status checker has failed 10 times in a row!",
+          type: "error",
+          time: 30000
+        });
+        step = "errored";
+        break;
+      }
+      await new Promise((r) => setTimeout(r, 1000));
+    }
+    if (step !== "modal-closing") {
+      // Update our watched list
+      const nid = notify({ text: "Fetching updated watched list.", type: "loading" });
+      try {
+        const w = await axios.get("/watched");
+        if (w?.data?.length > 0) {
+          watchedList.update((wl) => (wl = w.data));
+          notify({ id: nid, text: "Fetched updated watched list.", type: "success" });
+        }
+      } catch (err) {
+        console.error("jobWatcher: Getting updated watched list failed!", err);
+        notify({ id: nid, text: "Getting updated watched list failed!", type: "error" });
+      }
+    }
+  }
+
+  function modalClose() {
+    if (step === "job-running") {
+      notify({
+        text: "Sync will continue in the background.. please refresh the page periodically to view your updated list or come back later.",
+        time: 10000
+      });
+    }
+    step = "modal-closing";
+    onClose();
+  }
+
+  onMount(() => {
+    startJellyfinSync();
+  });
+
+  onDestroy(() => {
+    step = "starting";
+    jobId = undefined;
+    currentTask = undefined;
+    latestJobStatus = undefined;
+  });
+</script>
+
+<Modal title="Jellyfin Sync" maxWidth="700px" onClose={modalClose}>
+  <div class="ctr">
+    {#if step === "done"}
+      <Icon i="check" wh={60} />
+    {:else if step === "errored"}
+      <Icon i="close" wh={70} />
+    {:else}
+      <Spinner />
+    {/if}
+    <div>
+      {#if step === "starting"}
+        <h4 class="norm">Starting</h4>
+        <span>We are requesting a full sync</span>
+      {:else if step === "job-running"}
+        <h4 class="norm">Syncing</h4>
+        {#if currentTask}
+          <span>{currentTask}</span>
+        {/if}
+      {:else if step === "done"}
+        {#if !latestJobStatus?.errors || latestJobStatus?.errors?.length <= 0}
+          <h4 class="norm">Finished</h4>
+          <span>We have finished syncing. Looks like there were no errors!</span>
+        {:else}
+          <h4 class="norm">
+            Finished With {latestJobStatus?.errors?.length} Error{latestJobStatus?.errors
+              ?.length === 1
+              ? ""
+              : "s"}
+          </h4>
+          <span>Syncing has finished, but with errors:</span>
+          <ul>
+            {#each latestJobStatus?.errors as e}
+              <li>{e}</li>
+            {/each}
+          </ul>
+        {/if}
+      {:else if step === "errored"}
+        <h4 class="norm">We Errored!</h4>
+        <span>We errored before starting sync or the sync job was cancelled.</span>
+      {:else}
+        <h4 class="norm">Unknown State!</h4>
+        <span>We're not sure of the current sync status.</span>
+      {/if}
+    </div>
+  </div>
+</Modal>
+
+<style lang="scss">
+  .ctr {
+    display: flex;
+    flex-flow: row;
+    gap: 20px;
+    justify-content: start;
+    align-items: start;
+    margin-top: 25px;
+    margin-bottom: 15px;
+    margin-left: 15px;
+
+    & > div:last-of-type {
+      display: flex;
+      flex-flow: column;
+      gap: 8px;
+      padding: 8px 0;
+
+      & > span {
+        font-style: italic;
+
+        &::first-letter {
+          text-transform: uppercase;
+        }
+      }
+
+      & > ul {
+        padding-left: 25px;
+
+        li::first-letter {
+          text-transform: uppercase;
+        }
+      }
+    }
+  }
+</style>

src/types.ts

@@ -979,3 +979,21 @@ export enum GameWebsiteCategory {
   Steam = 13,
   Reddit = 14
 }
+
+export interface JellyfinSyncResponse {
+  jobId: string;
+}
+
+export enum JobStatus {
+  CREATED = "CREATED",
+  RUNNING = "RUNNING",
+  DONE = "DONE",
+  CANCELLED = "CANCELLED"
+}
+
+export interface GetJobResponse {
+  name: string;
+  status: JobStatus;
+  currentTask?: string;
+  errors: string[];
+}