fix: refactoring, adding express api.

src/features/api/upload/helpers.ts

@@ -1,145 +0,0 @@
-import {Backup, DatabaseWith} from "@/db/schema/07_database";
-import {dispatchStorage} from "@/features/storages/dispatch";
-import type {StorageInput, StorageResult} from "@/features/storages/types";
-import * as drizzleDb from "@/db";
-import {withUpdatedAt} from "@/db/utils";
-import {eq} from "drizzle-orm";
-import {db} from "@/db";
-import {PassThrough} from "stream";
-import fs from "node:fs";
-import forge from "node-forge";
-import crypto from "node:crypto";
-import {pipeline} from "node:stream";
-import { promisify } from "util";
-
-
-export async function storeBackupFilesStream(
-    backup: Backup,
-    database: DatabaseWith,
-    fileStream: NodeJS.ReadableStream,
-    fileName: string
-): Promise<StorageResult[]> {
-
-    const settings = await db.query.setting.findFirst({
-        where: eq(drizzleDb.schemas.setting.name, "system"),
-        with: {storageChannel: true},
-    });
-
-    const defaultPolicy = settings?.storageChannel
-        ? [{
-            id: null,
-            storageChannelId: settings.storageChannel.id,
-            enabled: settings.storageChannel.enabled,
-        }]
-        : [];
-
-    const enabledPolicies = database.storagePolicies?.filter(p => p.enabled) ?? [];
-    const policies = enabledPolicies.length ? enabledPolicies : defaultPolicy;
-
-    if (!policies.length) {
-        await db.update(drizzleDb.schemas.backup)
-            .set(withUpdatedAt({status: "failed"}))
-            .where(eq(drizzleDb.schemas.backup.id, backup.id));
-        return [];
-    }
-
-    const storagePath = `backups/${database.project?.slug}/${fileName}`;
-    // const teeStreams = teeStreamSafe(fileStream, policies.length);
-    const teeStreams = teeStream(fileStream, policies.length);
-
-
-    const results = await Promise.all(
-        policies.map(async (policy, index) => {
-            const pass = teeStreams[index];
-            console.log("policy", index, policy.id)
-            const [backupStorage] = await db
-                .insert(drizzleDb.schemas.backupStorage)
-                .values({
-                    backupId: backup.id,
-                    storageChannelId: policy.storageChannelId,
-                    status: "pending",
-                    path: storagePath,
-                })
-                .returning();
-
-            const input: StorageInput = {
-                action: "upload",
-                data: {
-                    path: storagePath,
-                    file: teeStreams[index],
-                },
-            };
-
-
-            let result: StorageResult;
-
-            try {
-                result = policy.id
-                    ? await dispatchStorage(input, policy.id)
-                    : await dispatchStorage(input, undefined, policy.storageChannelId);
-            } catch (err: any) {
-                pass.destroy(err);
-                result = {success: false, provider: null, error: err.message};
-            }
-
-            await db.update(drizzleDb.schemas.backupStorage)
-                .set(withUpdatedAt({status: result.success ? "success" : "failed"}))
-                .where(eq(drizzleDb.schemas.backupStorage.id, backupStorage.id));
-
-            return result;
-        })
-    );
-    console.log("result upload", results);
-
-    const backupStatus = results.some(r => r.success) ? "success" : "failed";
-
-    await db.update(drizzleDb.schemas.backup)
-        .set(withUpdatedAt({status: backupStatus}))
-        .where(eq(drizzleDb.schemas.backup.id, backup.id));
-
-    return results;
-}
-
-
-export function createDecryptionStream(
-    encryptedAesKeyHex: string,
-    ivHex: string
-) {
-    const privateKeyPem = fs.readFileSync(
-        "private/keys/server_private.pem",
-        "utf8"
-    );
-
-    const privateKey = forge.pki.privateKeyFromPem(privateKeyPem);
-
-    const encryptedBytes = forge.util.hexToBytes(encryptedAesKeyHex);
-    const aesKeyBytes = privateKey.decrypt(encryptedBytes, "RSA-OAEP", {
-        md: forge.md.sha256.create(),
-        mgf1: {md: forge.md.sha256.create()},
-    });
-
-    const aesKey = Buffer.from(aesKeyBytes, "binary");
-    const iv = Buffer.from(ivHex, "hex");
-
-    return crypto.createDecipheriv("aes-256-cbc", aesKey, iv);
-}
-
-
-export function getFileExtension(dbType: string) {
-    switch (dbType) {
-        case "postgresql":
-            return ".dump";
-        case "mysql":
-            return ".sql";
-        default:
-            return ".dump";
-    }
-}
-
-const pipelinePromise = promisify(pipeline);
-
-function teeStream(stream: NodeJS.ReadableStream, n: number): PassThrough[] {
-    const taps = Array.from({ length: n }, () => new PassThrough());
-    taps.forEach(tap => pipelinePromise(stream, tap).catch(err => tap.destroy(err)));
-    return taps;
-}

src/features/api/upload/helpers/common.ts

@@ -6,8 +6,10 @@ import {db} from "@/db";
 import * as drizzleDb from "@/db";
 import {eq} from "drizzle-orm";
 import {withUpdatedAt} from "@/db/utils";
-import {saveStreamToTempFile} from "@/features/api/upload/helpers/file";
-
+import {sendNotificationsBackupRestore} from "@/features/notifications/helpers";
+import {eventEmitter} from "@/features/shared/event";
+import forge from "node-forge";
+import crypto from "node:crypto";
 
 /**
  * Save stream to a temporary file and upload to all storage providers in parallel.
@@ -15,12 +17,11 @@ import {saveStreamToTempFile} from "@/features/api/upload/helpers/file";
 export default async function uploadTempFileToProviders(
     backup: Backup,
     database: DatabaseWith,
-    inputStream: NodeJS.ReadableStream,
+    // inputStream: NodeJS.ReadableStream,
+    tmpPath: string,
     fileName: string
 ): Promise<StorageResult[]> {
 
-    const tmpPath = await saveStreamToTempFile(inputStream, fileName);
-
     const stats = fs.statSync(tmpPath);
     const fileSize = stats.size;
     console.log(tmpPath);
@@ -78,8 +79,6 @@ export default async function uploadTempFileToProviders(
             } catch (err: any) {
                 result = {success: false, provider: null, error: err.message};
             }
-            console.log(result);
-
             await db.update(drizzleDb.schemas.backupStorage)
                 .set(withUpdatedAt({status: result.success ? "success" : "failed"}))
                 .where(eq(drizzleDb.schemas.backupStorage.id, backupStorage.id));
@@ -99,6 +98,36 @@ export default async function uploadTempFileToProviders(
     fs.existsSync(tmpPath) && fs.unlinkSync(tmpPath);
 
     console.log(results);
+    eventEmitter.emit('modification', {update: true});
+
+    if (!backupStatus) {
+        await sendNotificationsBackupRestore(database, "error_backup");
+    }
+    await sendNotificationsBackupRestore(database, "success_backup");
 
     return results;
 }
+
+
+export function createDecryptionStream(
+    encryptedAesKeyHex: string,
+    ivHex: string
+) {
+    const privateKeyPem = fs.readFileSync(
+        "private/keys/server_private.pem",
+        "utf8"
+    );
+
+    const privateKey = forge.pki.privateKeyFromPem(privateKeyPem);
+
+    const encryptedBytes = forge.util.hexToBytes(encryptedAesKeyHex);
+    const aesKeyBytes = privateKey.decrypt(encryptedBytes, "RSA-OAEP", {
+        md: forge.md.sha256.create(),
+        mgf1: {md: forge.md.sha256.create()},
+    });
+
+    const aesKey = Buffer.from(aesKeyBytes, "binary");
+    const iv = Buffer.from(ivHex, "hex");
+
+    return crypto.createDecipheriv("aes-256-cbc", aesKey, iv);
+}

src/features/api/upload/helpers/file.ts

@@ -13,3 +13,15 @@ export async function saveStreamToTempFile(stream: NodeJS.ReadableStream, fileNa
     await pipelineAsync(stream, writeStream);
     return tmpPath;
 }
+
+
+export function getFileExtension(dbType: string) {
+    switch (dbType) {
+        case "postgresql":
+            return ".dump";
+        case "mysql":
+            return ".sql";
+        default:
+            return ".dump";
+    }
+}

src/features/api/upload/index.ts

@@ -3,13 +3,13 @@ import {v4 as uuidv4} from "uuid";
 import {and, eq} from "drizzle-orm";
 import * as drizzleDb from "@/db";
 import {db} from "@/db";
-import {createDecryptionStream, getFileExtension} from "./helpers";
 import {isUuidv4} from "@/utils/verify-uuid";
-import uploadTempFileToProviders from "@/features/api/upload/helpers/common";
+import uploadTempFileToProviders, {createDecryptionStream} from "@/features/api/upload/helpers/common";
 import {sendNotificationsBackupRestore} from "@/features/notifications/helpers";
 import {Backup} from "@/db/schema/07_database";
 import {withUpdatedAt} from "@/db/utils";
 import {eventEmitter} from "@/features/shared/event";
+import {getFileExtension, saveStreamToTempFile} from "@/features/api/upload/helpers/file";
 
 const router: Router = express.Router();
 
@@ -86,9 +86,15 @@ router.post("/:agentId", async (req: Request, res: Response) => {
 
             const decryptedStream = req.pipe(decipher);
 
-            await uploadTempFileToProviders(backup, database, decryptedStream, fileName);
-            await sendNotificationsBackupRestore(database, "success_backup");
-            eventEmitter.emit('modification', {update: true});
+            const tmpPath = await saveStreamToTempFile(decryptedStream, fileName);
+
+            if (!tmpPath) {
+                return res.status(500).json({
+                    error: "Unable to save tmp backup file",
+                });
+            }
+
+            uploadTempFileToProviders(backup, database, tmpPath, fileName);
 
             return res.json({success: true});
         } else {

src/features/storages/helpers.ts

@@ -11,7 +11,7 @@ import * as drizzleDb from "@/db";
 import {withUpdatedAt} from "@/db/utils";
 import {eq} from "drizzle-orm";
 import {db} from "@/db";
-import crypto, {createHash} from "crypto";
+import {createHash} from "crypto";
 import {getServerUrl} from "@/utils/get-server-url";
 import path from "path";
 

src/features/storages/providers/google-drive/helpers.ts

@@ -1,36 +1,3 @@
-// import {drive_v3, google} from "googleapis";
-// import Drive = drive_v3.Drive;
-// import {GoogleDriveConfig} from "@/features/storages/providers/google-drive/types";
-//
-//
-// export async function getGoogleDriveClient(config: GoogleDriveConfig): Promise<Drive> {
-//     const auth = new google.auth.JWT({
-//         email: config.clientEmail,
-//         key: config.privateKey?.replace(/\\n/g, "\n"),
-//         scopes: ["https://www.googleapis.com/auth/drive"],
-//     });
-//
-//     return google.drive({
-//         version: "v3",
-//         auth,
-//     });
-// }
-//
-//
-// export async function findFileByName(
-//     drive: any,
-//     name: string,
-//     folderId: string
-// ): Promise<string | null> {
-//     const res = await drive.files.list({
-//         q: `name='${name}' and '${folderId}' in parents and trashed=false`,
-//         fields: "files(id)",
-//         pageSize: 1,
-//     });
-//
-//     return res.data.files?.[0]?.id ?? null;
-// }
-
 import {drive_v3, google} from "googleapis";
 import {GoogleDriveConfig} from "@/features/storages/providers/google-drive/types";
 import Drive = drive_v3.Drive;
@@ -42,7 +9,6 @@ export async function getGoogleDriveClient(config: GoogleDriveConfig): Promise<D
     const oauth2Client = new google.auth.OAuth2(
         config.clientId,
         config.clientSecret,
-        // config.redirectUri
         baseUrl
     );
 

src/features/storages/providers/google-drive/index.ts

@@ -77,8 +77,6 @@ export async function getGoogleDrive(
     const fileId = await resolveFilePath(client, input.data.path, config.folderId);
     if (!fileId) return {success: false, provider: "google-drive", error: "File not found"};
 
-
-
     const res = await client.files.get(
         {fileId, alt: "media", supportsAllDrives: true},
         {responseType: "stream"}

src/features/storages/providers/index.ts

@@ -1,7 +1,7 @@
 import {
     StorageProviderKind,
     StorageInput,
-    StorageResult, StorageMetaData,
+    StorageResult,
 } from '../types';
 
 import {uploadLocal, getLocal, deleteLocal, pingLocal} from './local';
@@ -39,8 +39,6 @@ const handlers: Record<StorageProviderKind, ProviderHandler> = {
         delete: deleteGoogleDrive,
         ping: pingGoogleDrive,
     }
-    // gcs: null as any,
-    // azure: null as any,
 };
 
 export async function dispatchViaProvider(

src/features/storages/providers/s3.ts

@@ -35,26 +35,22 @@ export async function uploadS3(
     input: { data: StorageUploadInput, metadata?: StorageMetaData }
 ): Promise<StorageResult> {
     const client = await getS3Client(config);
-    console.log(client);
     await ensureBucket(config);
 
     const key = `${BASE_DIR}${input.data.path}`;
     const file = input.data.file;
 
-    console.log(key)
-
     let uploadStream: Readable;
     if (Buffer.isBuffer(file) || file instanceof Uint8Array) {
         uploadStream = Readable.from(file);
     } else if ((file as any).pipe) {
         uploadStream = file;
     } else {
-        throw new Error("Unsupported file type for streaming upload");
+        return {success: false, provider: "s3", error: "Unsupported file type for streaming upload"};
     }
 
     try {
         const result = await client.putObject(config.bucketName, key, uploadStream, input.data.size);
-        console.log(result);
     } catch (err: any) {
         return {success: false, provider: "s3", error: err.message};
     }