Release 2.6.14

src/web/lib/media/upload-sync.js

@@ -453,6 +453,23 @@ function createUploadSyncService(options) {
     return Boolean(localUploadDir);
   }
 
+  async function fetchLivePlayerRegistrations() {
+    if (!pool || typeof fetchPlayerRegistrations !== 'function') {
+      return [];
+    }
+
+    try {
+      const players = await fetchPlayerRegistrations(pool);
+      return Array.isArray(players)
+        ? players.filter(function (player) {
+          return isRecentPlayerRegistration(player, 60);
+        })
+        : [];
+    } catch (_error) {
+      return [];
+    }
+  }
+
   function isPlayerUnavailableError(error) {
     const code = String(error && error.cause && error.cause.code || error && error.code || '').trim().toUpperCase();
     return code === 'ENOTFOUND' || code === 'ECONNREFUSED' || code === 'EAI_AGAIN' || code === 'ETIMEDOUT';
@@ -462,16 +479,33 @@ function createUploadSyncService(options) {
     return Boolean(response) && Number(response.status) === 503;
   }
 
+  function buildPendingPlayerUploadSyncKey(operation) {
+    const uploadPath = normalizeUploadReference(operation && operation.uploadPath);
+    const metadata = operation && operation.metadata && typeof operation.metadata === 'object'
+      ? operation.metadata
+      : null;
+    const playerIdentifier = String((operation && operation.playerIdentifier) || (metadata && metadata.playerIdentifier) || '').trim();
+    const playerInternalBaseUrl = normalizeBaseUrl((operation && operation.playerInternalBaseUrl) || (metadata && metadata.playerInternalBaseUrl) || '');
+
+    return [uploadPath, playerIdentifier, playerInternalBaseUrl].filter(Boolean).join('|');
+  }
+
   function queuePlayerUploadSync(operation) {
     if (!operation || !operation.uploadPath) {
       return;
     }
 
-    pendingPlayerUploadSyncs.set(normalizeUploadReference(operation.uploadPath), {
+    const metadata = operation.metadata && typeof operation.metadata === 'object' ? operation.metadata : null;
+    const playerIdentifier = String((operation && operation.playerIdentifier) || (metadata && metadata.playerIdentifier) || '').trim();
+    const playerInternalBaseUrl = normalizeBaseUrl((operation && operation.playerInternalBaseUrl) || (metadata && metadata.playerInternalBaseUrl) || '');
+
+    pendingPlayerUploadSyncs.set(buildPendingPlayerUploadSyncKey(operation), {
       type: operation.type === 'delete' ? 'delete' : 'put',
       uploadPath: normalizeUploadReference(operation.uploadPath),
       uploadDir: operation.uploadDir || null,
-      metadata: operation.metadata || null
+      metadata: metadata,
+      playerIdentifier: playerIdentifier || null,
+      playerInternalBaseUrl: playerInternalBaseUrl || null
     });
 
     schedulePendingPlayerUploadSyncFlush();
@@ -592,20 +626,26 @@ function createUploadSyncService(options) {
     }
   }
 
-  async function syncUploadRefsToPlayer(uploadRefs, localUploadDir) {
+  async function syncUploadRefsToPlayer(uploadRefs, localUploadDir, preferredPlayerIdentifier, preferredPlayerInternalBaseUrl) {
     if (!shouldMirrorUploads(localUploadDir)) {
       return;
     }
 
-    const resolvedPlayerInternalBaseUrl = await getPlayerInternalBaseUrl();
+    const resolvedPlayerInternalBaseUrl = await getPlayerInternalBaseUrl(preferredPlayerIdentifier, preferredPlayerInternalBaseUrl);
     const uniqueRefs = Array.from(new Set((uploadRefs || []).map(normalizeUploadReference).filter(Boolean)));
     for (let i = 0; i < uniqueRefs.length; i += 1) {
-      const success = await pushUploadFileToPlayer(uniqueRefs[i], localUploadDir, resolvedPlayerInternalBaseUrl);
+      const success = await pushUploadFileToPlayer(uniqueRefs[i], localUploadDir, resolvedPlayerInternalBaseUrl, preferredPlayerIdentifier);
       if (!success) {
         queuePlayerUploadSync({
           type: 'put',
           uploadPath: uniqueRefs[i],
-          uploadDir: localUploadDir
+          uploadDir: localUploadDir,
+          playerIdentifier: preferredPlayerIdentifier,
+          playerInternalBaseUrl: preferredPlayerInternalBaseUrl,
+          metadata: preferredPlayerIdentifier || preferredPlayerInternalBaseUrl ? {
+            playerIdentifier: preferredPlayerIdentifier || null,
+            playerInternalBaseUrl: preferredPlayerInternalBaseUrl || null
+          } : null
         });
       }
     }
@@ -723,10 +763,27 @@ function createUploadSyncService(options) {
       return;
     }
 
-    return queueMediaSyncTask('media-sync:' + operation.key, 'Media sync', {
-      mode: 'playlist',
-      operation: operation
-    });
+    const players = await fetchLivePlayerRegistrations();
+    if (!players.length) {
+      return queueMediaSyncTask('media-sync:' + operation.key, 'Media sync', {
+        mode: 'playlist',
+        operation: operation
+      });
+    }
+
+    return Promise.all(players.map(function (player) {
+      const playerIdentifier = String(player && player.identifier || '').trim();
+      const playerInternalBaseUrl = String(player && player.internal_base_url || '').trim().replace(/\/$/, '');
+      const playerPublicBaseUrl = String(player && player.public_base_url || '').trim().replace(/\/$/, '');
+
+      return queueMediaSyncTask('media-sync:' + operation.key + (playerIdentifier ? ':' + playerIdentifier : ''), 'Media sync', {
+        mode: 'playlist',
+        operation: operation,
+        playerIdentifier: playerIdentifier || null,
+        playerInternalBaseUrl: playerInternalBaseUrl || null,
+        playerPublicBaseUrl: playerPublicBaseUrl || null
+      });
+    }));
   }
 
   async function flushPendingPlaylistUploadSyncs() {
@@ -787,22 +844,27 @@ function createUploadSyncService(options) {
     }
 
     pendingPlayerUploadSyncFlushInFlight = (async function () {
-      const pendingEntries = Array.from(pendingPlayerUploadSyncs.values());
-      const playerMetadata = pendingEntries.length && pendingEntries[0] && pendingEntries[0].metadata
-        ? pendingEntries[0].metadata
-        : await getPlayerTaskMetadata();
-      if (playerMetadata && playerMetadata.playerActive === false) {
-        pendingPlayerUploadSyncs.clear();
-        pendingPlayerUploadSyncRetryLogAt = 0;
-        return;
-      }
-      const resolvedPlayerInternalBaseUrl = playerMetadata && playerMetadata.playerInternalBaseUrl
-        ? playerMetadata.playerInternalBaseUrl
-        : await getPlayerInternalBaseUrl(playerMetadata && playerMetadata.playerIdentifier, playerMetadata && playerMetadata.playerInternalBaseUrl);
+      const pendingEntries = Array.from(pendingPlayerUploadSyncs.entries());
+      const firstOperation = pendingEntries.length && pendingEntries[0] ? pendingEntries[0][1] : null;
+      const summaryPlayerMetadata = firstOperation && firstOperation.metadata
+        ? firstOperation.metadata
+        : await getPlayerTaskMetadata(firstOperation && firstOperation.playerIdentifier, firstOperation && firstOperation.playerInternalBaseUrl);
       let successCount = 0;
       let failureCount = 0;
       for (let i = 0; i < pendingEntries.length; i += 1) {
-        const operation = pendingEntries[i];
+        const entry = pendingEntries[i];
+        const pendingKey = entry[0];
+        const operation = entry[1];
+        const playerMetadata = operation && operation.metadata
+          ? operation.metadata
+          : await getPlayerTaskMetadata(operation && operation.playerIdentifier, operation && operation.playerInternalBaseUrl);
+        if (playerMetadata && playerMetadata.playerActive === false) {
+          pendingPlayerUploadSyncs.delete(pendingKey);
+          continue;
+        }
+        const resolvedPlayerInternalBaseUrl = playerMetadata && playerMetadata.playerInternalBaseUrl
+          ? playerMetadata.playerInternalBaseUrl
+          : await getPlayerInternalBaseUrl(playerMetadata && playerMetadata.playerIdentifier, playerMetadata && playerMetadata.playerInternalBaseUrl);
         let success = false;
         if (operation.type === 'delete') {
           success = await removeUploadFileFromPlayer(operation.uploadPath, operation.uploadDir, resolvedPlayerInternalBaseUrl, playerMetadata && playerMetadata.playerIdentifier);
@@ -811,20 +873,20 @@ function createUploadSyncService(options) {
         }
         if (success) {
           successCount += 1;
-          pendingPlayerUploadSyncs.delete(operation.uploadPath);
+          pendingPlayerUploadSyncs.delete(pendingKey);
         } else {
           failureCount += 1;
         }
       }
 
       if (successCount) {
-        logMediaSyncSummary('info', `Media sync completed ${successCount} upload${successCount === 1 ? '' : 's'}`, playerMetadata);
+        logMediaSyncSummary('info', `Media sync completed ${successCount} upload${successCount === 1 ? '' : 's'}`, summaryPlayerMetadata);
       }
       if (failureCount) {
         const now = Date.now();
         if (!pendingPlayerUploadSyncRetryLogAt || now - pendingPlayerUploadSyncRetryLogAt >= PLAYER_UPLOAD_SYNC_RETRY_LOG_INTERVAL_MS) {
           pendingPlayerUploadSyncRetryLogAt = now;
-          logMediaSyncSummary('warn', `Player unavailable, retry queued for ${failureCount} upload${failureCount === 1 ? '' : 's'}`, playerMetadata);
+          logMediaSyncSummary('warn', `Player unavailable, retry queued for ${failureCount} upload${failureCount === 1 ? '' : 's'}`, summaryPlayerMetadata);
         }
       } else if (!pendingPlayerUploadSyncs.size) {
         pendingPlayerUploadSyncRetryLogAt = 0;
@@ -892,7 +954,7 @@ function createUploadSyncService(options) {
       const operation = normalizePlaylistUploadSyncOperation(taskPayload.operation || taskPayload);
 
       if (operation.nextUploadRefs.length) {
-        await syncUploadRefsToPlayer(operation.nextUploadRefs, operation.localUploadDir);
+        await syncUploadRefsToPlayer(operation.nextUploadRefs, operation.localUploadDir, taskPayload.playerIdentifier, taskPayload.playerInternalBaseUrl);
       }
 
       if (operation.previousUploadRefs.length) {