fix(backup): attach reconnecting restore subscriptions to the in-flight job instead of re-running the restore

apps/dokploy/server/api/routers/backup.ts

@@ -79,6 +79,15 @@ interface RcloneFile {
 	};
 }
 
+interface RestoreJob {
+	logs: string[];
+	done: boolean;
+}
+
+// The websocket client re-subscribes on reconnect; attaching to the in-flight
+// job instead of starting a new restore prevents concurrent duplicate runs.
+const activeRestores = new Map<string, RestoreJob>();
+
 export const backupRouter = createTRPCRouter({
 	create: protectedProcedure
 		.input(apiCreateBackup)
@@ -580,46 +589,70 @@ export const backupRouter = createTRPCRouter({
 				});
 			}
 			const destination = await findDestinationById(input.destinationId);
-			const queue: string[] = [];
-			let done = false;
-			const onLog = (log: string) => queue.push(log);
-			const runRestore = async () => {
-				if (input.backupType === "database") {
-					if (input.databaseType === "postgres") {
-						const postgres = await findPostgresById(input.databaseId);
-						await restorePostgresBackup(postgres, destination, input, onLog);
-					} else if (input.databaseType === "mysql") {
-						const mysql = await findMySqlById(input.databaseId);
-						await restoreMySqlBackup(mysql, destination, input, onLog);
-					} else if (input.databaseType === "mariadb") {
-						const mariadb = await findMariadbById(input.databaseId);
-						await restoreMariadbBackup(mariadb, destination, input, onLog);
-					} else if (input.databaseType === "mongo") {
-						const mongo = await findMongoById(input.databaseId);
-						await restoreMongoBackup(mongo, destination, input, onLog);
-					} else if (input.databaseType === "libsql") {
-						const libsql = await findLibsqlById(input.databaseId);
-						await restoreLibsqlBackup(libsql, destination, input, onLog);
-					} else if (input.databaseType === "web-server") {
-						await restoreWebServerBackup(destination, input.backupFile, onLog);
+			const jobKey = [
+				input.backupType,
+				input.databaseId,
+				input.databaseType,
+				input.databaseName,
+				input.backupFile,
+				input.destinationId,
+			].join("|");
+
+			let job = activeRestores.get(jobKey);
+			if (!job) {
+				const newJob: RestoreJob = { logs: [], done: false };
+				activeRestores.set(jobKey, newJob);
+				job = newJob;
+				const onLog = (log: string) => newJob.logs.push(log);
+				const runRestore = async () => {
+					if (input.backupType === "database") {
+						if (input.databaseType === "postgres") {
+							const postgres = await findPostgresById(input.databaseId);
+							await restorePostgresBackup(postgres, destination, input, onLog);
+						} else if (input.databaseType === "mysql") {
+							const mysql = await findMySqlById(input.databaseId);
+							await restoreMySqlBackup(mysql, destination, input, onLog);
+						} else if (input.databaseType === "mariadb") {
+							const mariadb = await findMariadbById(input.databaseId);
+							await restoreMariadbBackup(mariadb, destination, input, onLog);
+						} else if (input.databaseType === "mongo") {
+							const mongo = await findMongoById(input.databaseId);
+							await restoreMongoBackup(mongo, destination, input, onLog);
+						} else if (input.databaseType === "libsql") {
+							const libsql = await findLibsqlById(input.databaseId);
+							await restoreLibsqlBackup(libsql, destination, input, onLog);
+						} else if (input.databaseType === "web-server") {
+							await restoreWebServerBackup(
+								destination,
+								input.backupFile,
+								onLog,
+							);
+						}
+					} else if (input.backupType === "compose") {
+						const compose = await findComposeById(input.databaseId);
+						await restoreComposeBackup(compose, destination, input, onLog);
 					}
-				} else if (input.backupType === "compose") {
-					const compose = await findComposeById(input.databaseId);
-					await restoreComposeBackup(compose, destination, input, onLog);
-				}
-			};
-			runRestore()
-				.catch((error) => {
-					onLog(
-						`Error: ${error instanceof Error ? error.message : String(error)}`,
-					);
-				})
-				.finally(() => {
-					done = true;
-				});
-			while (!done || queue.length > 0) {
-				if (queue.length > 0) {
-					yield queue.shift()!;
+				};
+				runRestore()
+					.catch((error) => {
+						onLog(
+							`Error: ${error instanceof Error ? error.message : String(error)}`,
+						);
+					})
+					.finally(() => {
+						newJob.done = true;
+						setTimeout(() => {
+							if (activeRestores.get(jobKey) === newJob) {
+								activeRestores.delete(jobKey);
+							}
+						}, 5000);
+					});
+			}
+
+			let cursor = 0;
+			while (!job.done || cursor < job.logs.length) {
+				if (cursor < job.logs.length) {
+					yield job.logs[cursor++]!;
 				} else {
 					await new Promise((r) => setTimeout(r, 50));
 				}