Playback tracking rework

This rework looks for existing playback records based on the following criteria:
activity matching `NowPlayingItemId` within the last 1 hour with Playback progress <=80%

if theres a matching record:
update the PlaybackDuration to be existing PlaybackDuration + new PlaybackDuration progress

Update ActivityDateInserted to current time

Related issues: #56 #116

backend/migrations/058_jf_recent_playback_activity_function.js

@@ -0,0 +1,88 @@
+exports.up = async function(knex) {
+  try
+  {
+     await knex.schema.raw(`
+     CREATE OR REPLACE FUNCTION jf_recent_playback_activity(hour_offset INT)
+      RETURNS TABLE (
+          "RunTimeTicks" BIGINT,
+          "Progress" NUMERIC,
+          "Id" TEXT,
+          "IsPaused" BOOLEAN,
+          "UserId" TEXT,
+          "UserName" TEXT,
+          "Client" TEXT,
+          "DeviceName" TEXT,
+          "DeviceId" TEXT,
+          "ApplicationVersion" TEXT,
+          "NowPlayingItemId" TEXT,
+          "NowPlayingItemName" TEXT,
+          "SeasonId" TEXT,
+          "SeriesName" TEXT,
+          "EpisodeId" TEXT,
+          "PlaybackDuration" BIGINT,
+          "ActivityDateInserted" timestamptz,
+          "PlayMethod" TEXT,
+          "MediaStreams" JSON,
+          "TranscodingInfo" JSON,
+          "PlayState" JSON,
+          "OriginalContainer" TEXT,
+          "RemoteEndPoint" TEXT,
+          "ServerId" TEXT,
+          "Imported" BOOLEAN,
+          "RowNum" BIGINT
+      )
+      AS $$
+      BEGIN
+          RETURN QUERY
+          WITH rankedactivities AS (
+               SELECT COALESCE(i."RunTimeTicks", e."RunTimeTicks") AS "RunTimeTicks",
+                  (a."PlaybackDuration"::numeric(100,0) / COALESCE(i."RunTimeTicks"::numeric(100,0), e."RunTimeTicks"::numeric(100,0), 1.0) * 100::numeric)::numeric(10,2) AS "Progress",
+                  a."Id",
+                  a."IsPaused",
+                  a."UserId",
+                  a."UserName",
+                  a."Client",
+                  a."DeviceName",
+                  a."DeviceId",
+                  a."ApplicationVersion",
+                  a."NowPlayingItemId",
+                  a."NowPlayingItemName",
+                  a."SeasonId",
+                  a."SeriesName",
+                  a."EpisodeId",
+                  a."PlaybackDuration",
+                  a."ActivityDateInserted",
+                  a."PlayMethod",
+                  a."MediaStreams",
+                  a."TranscodingInfo",
+                  a."PlayState",
+                  a."OriginalContainer",
+                  a."RemoteEndPoint",
+                  a."ServerId",
+                  a.imported,
+                  row_number() OVER (PARTITION BY a."NowPlayingItemId" ORDER BY a."ActivityDateInserted" DESC) AS rownum
+                 FROM jf_playback_activity a
+                   LEFT JOIN jf_library_items i ON a."NowPlayingItemId" = i."Id"
+                   LEFT JOIN jf_library_episodes e ON a."EpisodeId" = e."EpisodeId"
+                WHERE a."ActivityDateInserted" > (CURRENT_TIMESTAMP - (hour_offset || ' hours')::interval)
+                ORDER BY a."ActivityDateInserted" DESC
+              )
+           SELECT * FROM rankedactivities WHERE rankedactivities.rownum = 1;
+      END;
+      $$ LANGUAGE plpgsql;
+          
+     `
+           );
+
+}catch (error) {
+  console.error(error);
+}
+};
+
+exports.down = async function(knex) {
+  try {
+    await knex.raw(`DROP FUNCTION jf_recent_playback_activity;`);
+  } catch (error) {
+    console.error(error);
+  }
+};

backend/models/bulk_insert_update_handler.js

@@ -8,7 +8,7 @@
     {table:'jf_library_items',query:' ON CONFLICT ("Id") DO UPDATE SET "Name" = EXCLUDED."Name", "PremiereDate" = EXCLUDED."PremiereDate", "EndDate" = EXCLUDED."EndDate", "CommunityRating" = EXCLUDED."CommunityRating", "RunTimeTicks" = EXCLUDED."RunTimeTicks", "ProductionYear" = EXCLUDED."ProductionYear", "Type" = EXCLUDED."Type", "Status" = EXCLUDED."Status", "ImageTagsPrimary" = EXCLUDED."ImageTagsPrimary", "ImageTagsBanner" = EXCLUDED."ImageTagsBanner", "ImageTagsLogo" = EXCLUDED."ImageTagsLogo", "ImageTagsThumb" = EXCLUDED."ImageTagsThumb", "BackdropImageTags" = EXCLUDED."BackdropImageTags", "ParentId" = EXCLUDED."ParentId", "PrimaryImageHash" = EXCLUDED."PrimaryImageHash", archived=false'},
     {table:'jf_library_seasons',query:' ON CONFLICT ("Id") DO UPDATE SET "Name" = EXCLUDED."Name", "ParentLogoItemId" = EXCLUDED."ParentLogoItemId", "ParentBackdropItemId" = EXCLUDED."ParentBackdropItemId", "ParentBackdropImageTags" = EXCLUDED."ParentBackdropImageTags", "SeriesPrimaryImageTag" = EXCLUDED."SeriesPrimaryImageTag"'},
     {table:'jf_logging',query:` ON CONFLICT ("Id") DO UPDATE SET "Duration" = EXCLUDED."Duration", "Log"=EXCLUDED."Log", "Result"=EXCLUDED."Result" WHERE "jf_logging"."Result"='Running'`},
-    {table:'jf_playback_activity',query:' ON CONFLICT DO NOTHING'},
+    {table:'jf_playback_activity',query:' ON CONFLICT ("Id") DO UPDATE SET "PlaybackDuration" = EXCLUDED."PlaybackDuration", "ActivityDateInserted" = EXCLUDED."ActivityDateInserted"'},
     {table:'jf_playback_reporting_plugin_data',query:' ON CONFLICT DO NOTHING'},
     {table:'jf_users',query:' ON CONFLICT ("Id") DO UPDATE SET "Name" = EXCLUDED."Name", "PrimaryImageTag" = EXCLUDED."PrimaryImageTag", "LastLoginDate" = EXCLUDED."LastLoginDate", "LastActivityDate" = EXCLUDED."LastActivityDate"'}
   ]; 

backend/tasks/ActivityMonitor.js

@@ -41,6 +41,7 @@ async function ActivityMonitor(interval) {
 
       let WatchdogDataToInsert = [];
       let WatchdogDataToUpdate = [];
+   
       //filter fix if table is empty
 
       if (WatchdogData.length === 0) {
@@ -67,10 +68,10 @@ async function ActivityMonitor(interval) {
 
 
   
-      if (WatchdogDataToInsert.length !== 0) {
+      if (WatchdogDataToInsert.length > 0) {
+        //insert new rows where not existing items
         db.insertBulk("jf_activity_watchdog",WatchdogDataToInsert,jf_activity_watchdog_columns);
       }
-     
 
       //update wd state
       if(WatchdogDataToUpdate.length>0)
@@ -136,7 +137,7 @@ async function ActivityMonitor(interval) {
       const playbackData =  WatchdogData.filter((id) => !SessionData.some((row) => row.Id === id.Id));
 
       
-      const playbackToInsert = playbackData.map(obj => {
+      let playbackToInsert = playbackData.map(obj => {
         const uuid = randomUUID()
 
         obj.Id=uuid;
@@ -158,6 +159,35 @@ async function ActivityMonitor(interval) {
         return { ...rest };
       });
 
+    
+    const playbackToInsertIds=playbackToInsert.map((row) => row.NowPlayingItemId);
+
+    /////get data from jf_playback_activity within the last hour with progress of <=80% for current items in session
+    const ExistingRecords=await db.query(`SELECT * FROM jf_recent_playback_activity(1)`).then((res) => res.rows.filter((row) => playbackToInsertIds.includes(row.NowPlayingItemId) && row.Progress<=80.0));
+    let ExistingDataToUpdate = [];
+
+    //for each item in playbackToInsert, check if it exists in the recent playback activity and update accordingly
+      if(playbackToInsert.length>0 && ExistingRecords.length>0)
+      {
+   
+        ExistingDataToUpdate=playbackToInsert.filter((playbackData) => {
+          const existingrow=ExistingRecords.find((existing) => existing.NowPlayingItemId === playbackData.NowPlayingItemId);
+ 
+          if(existingrow)
+          {
+            playbackData.Id=existingrow.Id;
+            playbackData.PlaybackDuration=Number(existingrow.PlaybackDuration)+Number(playbackData.PlaybackDuration);
+            playbackData.ActivityDateInserted= moment().format('YYYY-MM-DD HH:mm:ss.SSSZ');
+            return true;
+          }
+          return false;
+        });
+      }
+
+      //remove items from playbackToInsert that already exists in the recent playback activity so it doesnt duplicate
+      playbackToInsert=playbackToInsert.filter((pb)=> !ExistingRecords.map(er=>er.NowPlayingItemId).includes(pb.NowPlayingItemId));
+
+
 
 
       if(toDeleteIds.length>0)
@@ -169,6 +199,11 @@ async function ActivityMonitor(interval) {
         await db.insertBulk('jf_playback_activity',playbackToInsert,columnsPlayback);
       }
 
+      if(ExistingDataToUpdate.length>0)
+      {
+        await db.insertBulk('jf_playback_activity',ExistingDataToUpdate,columnsPlayback);
+      }
+
 
       ///////////////////////////