feat: add docker pipeline and preview fixes
This commit is contained in:
@@ -0,0 +1,500 @@
|
||||
import {
|
||||
buildContentRange,
|
||||
buildVolumeMap,
|
||||
ensurePlainZipVolume,
|
||||
listCachedVolumeNames,
|
||||
parseSingleRange,
|
||||
readCachedZipEntryBytes,
|
||||
selectSegmentsForRange
|
||||
} from './modules/media-core.js';
|
||||
|
||||
const STATUS_CHANNEL_NAME = 'duplicati-media-status';
|
||||
const SEGMENT_CACHE_LIMIT = 64;
|
||||
const SESSION_PATH_PREFIX = '/__media__/session/';
|
||||
|
||||
const sessions = new Map();
|
||||
const statusChannel = typeof BroadcastChannel === 'function' ? new BroadcastChannel(STATUS_CHANNEL_NAME) : null;
|
||||
|
||||
function createAbortError() {
|
||||
const error = new Error('The active media request was aborted.');
|
||||
error.code = 'MEDIA_REQUEST_ABORTED';
|
||||
error.name = 'AbortError';
|
||||
return error;
|
||||
}
|
||||
|
||||
function throwIfAborted(signal) {
|
||||
if (!signal?.aborted) {
|
||||
return;
|
||||
}
|
||||
|
||||
throw createAbortError();
|
||||
}
|
||||
|
||||
function isAbortError(error) {
|
||||
return error?.name === 'AbortError' || error?.code === 'MEDIA_REQUEST_ABORTED';
|
||||
}
|
||||
|
||||
function emitStatus(session) {
|
||||
if (!statusChannel) {
|
||||
return;
|
||||
}
|
||||
|
||||
statusChannel.postMessage({
|
||||
type: 'status',
|
||||
sessionId: session.id,
|
||||
status: {
|
||||
phase: session.phase,
|
||||
currentVolume: session.currentVolume,
|
||||
cachedVolumes: [...session.cachedVolumes].sort(),
|
||||
lastError: session.lastError,
|
||||
registeredAt: session.registeredAt,
|
||||
requestVersion: session.requestVersion
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function setSessionStatus(session, patch, { requestVersion = null, force = false } = {}) {
|
||||
if (!force && requestVersion !== null && requestVersion !== session.requestVersion) {
|
||||
return;
|
||||
}
|
||||
|
||||
Object.assign(session, patch);
|
||||
emitStatus(session);
|
||||
}
|
||||
|
||||
function snapshotSession(session) {
|
||||
return {
|
||||
phase: session.phase,
|
||||
currentVolume: session.currentVolume,
|
||||
cachedVolumes: [...session.cachedVolumes].sort(),
|
||||
lastError: session.lastError,
|
||||
registeredAt: session.registeredAt,
|
||||
requestVersion: session.requestVersion
|
||||
};
|
||||
}
|
||||
|
||||
function createHeaders(session, contentLength) {
|
||||
return {
|
||||
'Accept-Ranges': 'bytes',
|
||||
'Content-Type': session.fileInfo.file.mime || 'application/octet-stream',
|
||||
'Content-Length': String(contentLength),
|
||||
'Cache-Control': 'no-store'
|
||||
};
|
||||
}
|
||||
|
||||
function cacheSegment(session, key, bytes) {
|
||||
if (session.segmentCache.has(key)) {
|
||||
session.segmentCache.delete(key);
|
||||
}
|
||||
|
||||
session.segmentCache.set(key, bytes);
|
||||
|
||||
while (session.segmentCache.size > SEGMENT_CACHE_LIMIT) {
|
||||
const oldestKey = session.segmentCache.keys().next().value;
|
||||
session.segmentCache.delete(oldestKey);
|
||||
}
|
||||
}
|
||||
|
||||
function trackController(session, controller) {
|
||||
session.activeControllers.add(controller);
|
||||
controller.signal.addEventListener(
|
||||
'abort',
|
||||
() => {
|
||||
session.activeControllers.delete(controller);
|
||||
},
|
||||
{ once: true }
|
||||
);
|
||||
return controller;
|
||||
}
|
||||
|
||||
function abortActiveWork(session) {
|
||||
for (const controller of session.activeControllers) {
|
||||
controller.abort();
|
||||
}
|
||||
|
||||
session.activeControllers.clear();
|
||||
session.volumeInflight = new Map();
|
||||
session.segmentInflight = new Map();
|
||||
}
|
||||
|
||||
function createRequestController(session, request) {
|
||||
const controller = trackController(session, new AbortController());
|
||||
|
||||
if (request.signal) {
|
||||
if (request.signal.aborted) {
|
||||
controller.abort();
|
||||
} else {
|
||||
request.signal.addEventListener(
|
||||
'abort',
|
||||
() => {
|
||||
controller.abort();
|
||||
},
|
||||
{ once: true }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return controller;
|
||||
}
|
||||
|
||||
async function ensureSessionVolume(session, volume, requestVersion, signal) {
|
||||
throwIfAborted(signal);
|
||||
const cacheKey = `${session.sourceId}::${volume.name}`;
|
||||
const inflightMap = session.volumeInflight;
|
||||
if (inflightMap.has(cacheKey)) {
|
||||
return inflightMap.get(cacheKey);
|
||||
}
|
||||
|
||||
const promise = ensurePlainZipVolume({
|
||||
sourceId: session.sourceId,
|
||||
volume,
|
||||
secrets: session.secrets,
|
||||
signal,
|
||||
onStatus: ({ phase, currentVolume }) => {
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase,
|
||||
currentVolume: currentVolume ?? volume.name
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
}
|
||||
})
|
||||
.then(({ cached }) => {
|
||||
throwIfAborted(signal);
|
||||
session.cachedVolumes.add(volume.name);
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: cached ? 'cache-hit' : 'ready',
|
||||
currentVolume: volume.name,
|
||||
lastError: null
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
})
|
||||
.catch((error) => {
|
||||
if (!isAbortError(error)) {
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: 'error',
|
||||
currentVolume: volume.name,
|
||||
lastError: error.message
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
}
|
||||
|
||||
throw error;
|
||||
})
|
||||
.finally(() => {
|
||||
inflightMap.delete(cacheKey);
|
||||
});
|
||||
|
||||
inflightMap.set(cacheKey, promise);
|
||||
return promise;
|
||||
}
|
||||
|
||||
async function restoreSegmentBytesForSession(session, segment, requestVersion, signal) {
|
||||
throwIfAborted(signal);
|
||||
const cacheKey = `${segment.segmentIndex}`;
|
||||
if (session.segmentCache.has(cacheKey)) {
|
||||
const bytes = session.segmentCache.get(cacheKey);
|
||||
session.segmentCache.delete(cacheKey);
|
||||
session.segmentCache.set(cacheKey, bytes);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
const inflightMap = session.segmentInflight;
|
||||
if (inflightMap.has(cacheKey)) {
|
||||
return inflightMap.get(cacheKey);
|
||||
}
|
||||
|
||||
const promise = (async () => {
|
||||
const volume = session.volumeByRef.get(segment.volumeRef);
|
||||
if (!volume) {
|
||||
throw new Error(`Missing volume metadata for ${segment.volumeRef}.`);
|
||||
}
|
||||
|
||||
await ensureSessionVolume(session, volume, requestVersion, signal);
|
||||
throwIfAborted(signal);
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: 'serving',
|
||||
currentVolume: volume.name
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
|
||||
const bytes = await readCachedZipEntryBytes({
|
||||
sourceId: session.sourceId,
|
||||
volumeName: volume.name,
|
||||
zip: segment.zip,
|
||||
signal
|
||||
});
|
||||
throwIfAborted(signal);
|
||||
|
||||
cacheSegment(session, cacheKey, bytes);
|
||||
return bytes;
|
||||
})().finally(() => {
|
||||
inflightMap.delete(cacheKey);
|
||||
});
|
||||
|
||||
inflightMap.set(cacheKey, promise);
|
||||
return promise;
|
||||
}
|
||||
|
||||
function createRangeStream(session, rangeInfo, requestVersion, requestController) {
|
||||
const overlaps = selectSegmentsForRange(session.fileInfo.segments, rangeInfo.start, rangeInfo.endExclusive);
|
||||
|
||||
return new ReadableStream({
|
||||
async start(controller) {
|
||||
try {
|
||||
for (const overlap of overlaps) {
|
||||
throwIfAborted(requestController.signal);
|
||||
const segmentBytes = await restoreSegmentBytesForSession(
|
||||
session,
|
||||
overlap.segment,
|
||||
requestVersion,
|
||||
requestController.signal
|
||||
);
|
||||
throwIfAborted(requestController.signal);
|
||||
controller.enqueue(
|
||||
segmentBytes.subarray(overlap.offsetWithinSegmentStart, overlap.offsetWithinSegmentEnd)
|
||||
);
|
||||
}
|
||||
|
||||
controller.close();
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: 'ready',
|
||||
currentVolume: null,
|
||||
lastError: null
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
} catch (error) {
|
||||
if (!isAbortError(error)) {
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: 'error',
|
||||
currentVolume: session.currentVolume,
|
||||
lastError: error.message
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
}
|
||||
|
||||
controller.error(error);
|
||||
} finally {
|
||||
session.activeControllers.delete(requestController);
|
||||
}
|
||||
},
|
||||
cancel() {
|
||||
requestController.abort();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function jsonResponse(payload, status = 200) {
|
||||
return new Response(JSON.stringify(payload), {
|
||||
status,
|
||||
headers: {
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function handleMessage(event) {
|
||||
const { data, ports } = event;
|
||||
const replyPort = ports?.[0] ?? null;
|
||||
if (!data?.type) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
if (data.type === 'register-session') {
|
||||
const session = {
|
||||
id: data.sessionId,
|
||||
sourceId: data.sourceId,
|
||||
fileInfo: data.fileInfo,
|
||||
secrets: data.secrets,
|
||||
volumeByRef: buildVolumeMap(data.fileInfo?.volumes ?? []),
|
||||
cachedVolumes: new Set(await listCachedVolumeNames(data.sourceId)),
|
||||
segmentCache: new Map(),
|
||||
segmentInflight: new Map(),
|
||||
volumeInflight: new Map(),
|
||||
activeControllers: new Set(),
|
||||
requestVersion: 0,
|
||||
phase: 'registered',
|
||||
currentVolume: null,
|
||||
lastError: null,
|
||||
registeredAt: new Date().toISOString()
|
||||
};
|
||||
|
||||
const previous = sessions.get(data.sessionId);
|
||||
if (previous) {
|
||||
abortActiveWork(previous);
|
||||
}
|
||||
|
||||
sessions.set(data.sessionId, session);
|
||||
emitStatus(session);
|
||||
replyPort?.postMessage({
|
||||
ok: true,
|
||||
status: snapshotSession(session)
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (data.type === 'unregister-session') {
|
||||
const session = sessions.get(data.sessionId);
|
||||
if (session) {
|
||||
abortActiveWork(session);
|
||||
}
|
||||
sessions.delete(data.sessionId);
|
||||
replyPort?.postMessage({ ok: true });
|
||||
return;
|
||||
}
|
||||
|
||||
if (data.type === 'get-session-status') {
|
||||
const session = sessions.get(data.sessionId);
|
||||
if (!session) {
|
||||
replyPort?.postMessage({
|
||||
ok: false,
|
||||
error: 'MEDIA_SESSION_NOT_FOUND'
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
replyPort?.postMessage({
|
||||
ok: true,
|
||||
status: snapshotSession(session)
|
||||
});
|
||||
return;
|
||||
}
|
||||
} catch (error) {
|
||||
replyPort?.postMessage({
|
||||
ok: false,
|
||||
error: error.message
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function handleMediaRequest(request) {
|
||||
const url = new URL(request.url);
|
||||
const sessionId = decodeURIComponent(url.pathname.slice(SESSION_PATH_PREFIX.length));
|
||||
const session = sessions.get(sessionId);
|
||||
|
||||
if (!session) {
|
||||
return jsonResponse(
|
||||
{
|
||||
error: {
|
||||
code: 'MEDIA_SESSION_NOT_FOUND',
|
||||
message: 'The requested media session is not registered.'
|
||||
}
|
||||
},
|
||||
404
|
||||
);
|
||||
}
|
||||
|
||||
const totalSize = Number(session.fileInfo.file.size ?? 0);
|
||||
if (!Number.isFinite(totalSize) || totalSize < 0) {
|
||||
return jsonResponse(
|
||||
{
|
||||
error: {
|
||||
code: 'MEDIA_SIZE_INVALID',
|
||||
message: 'The registered media session did not include a valid file size.'
|
||||
}
|
||||
},
|
||||
500
|
||||
);
|
||||
}
|
||||
|
||||
let rangeInfo;
|
||||
try {
|
||||
rangeInfo = parseSingleRange(request.headers.get('range'), totalSize);
|
||||
} catch (error) {
|
||||
return new Response(null, {
|
||||
status: 416,
|
||||
headers: {
|
||||
'Content-Range': `bytes */${totalSize}`
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
const headers = createHeaders(session, rangeInfo.contentLength);
|
||||
if (request.method === 'HEAD') {
|
||||
return new Response(null, {
|
||||
status: rangeInfo.partial ? 206 : 200,
|
||||
headers: {
|
||||
...headers,
|
||||
...(rangeInfo.partial
|
||||
? {
|
||||
'Content-Range': buildContentRange(rangeInfo.start, rangeInfo.endInclusive, totalSize)
|
||||
}
|
||||
: {})
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
abortActiveWork(session);
|
||||
session.requestVersion += 1;
|
||||
const requestVersion = session.requestVersion;
|
||||
const requestController = createRequestController(session, request);
|
||||
setSessionStatus(
|
||||
session,
|
||||
{
|
||||
phase: 'seeking',
|
||||
currentVolume: null,
|
||||
lastError: null
|
||||
},
|
||||
{ requestVersion }
|
||||
);
|
||||
|
||||
return new Response(createRangeStream(session, rangeInfo, requestVersion, requestController), {
|
||||
status: rangeInfo.partial ? 206 : 200,
|
||||
headers: {
|
||||
...headers,
|
||||
...(rangeInfo.partial
|
||||
? {
|
||||
'Content-Range': buildContentRange(rangeInfo.start, rangeInfo.endInclusive, totalSize)
|
||||
}
|
||||
: {})
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
self.addEventListener('install', (event) => {
|
||||
event.waitUntil(self.skipWaiting());
|
||||
});
|
||||
|
||||
self.addEventListener('activate', (event) => {
|
||||
event.waitUntil(self.clients.claim());
|
||||
});
|
||||
|
||||
self.addEventListener('message', (event) => {
|
||||
event.waitUntil(handleMessage(event));
|
||||
});
|
||||
|
||||
self.addEventListener('fetch', (event) => {
|
||||
const url = new URL(event.request.url);
|
||||
if (url.origin !== self.location.origin) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!url.pathname.startsWith(SESSION_PATH_PREFIX)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.request.method !== 'GET' && event.request.method !== 'HEAD') {
|
||||
event.respondWith(new Response(null, { status: 405 }));
|
||||
return;
|
||||
}
|
||||
|
||||
event.respondWith(handleMediaRequest(event.request));
|
||||
});
|
||||
Reference in New Issue
Block a user