chore(release): 1.1.2-rc.2

fix/s3-region

.github/workflows/docker.yml

@@ -26,76 +26,92 @@ on:
         required: true
 
 jobs:
-  build-and-push:
-    runs-on: ubuntu-latest
-    permissions:
-      packages: write
-      contents: read
-
+  build:
+    name: Build and push Docker images
+    runs-on: ${{ matrix.platform == 'linux/amd64' && 'ubuntu-latest' || matrix.platform == 'linux/arm64' && 'ubuntu-24.04-arm' }}
+    strategy:
+      fail-fast: false
+      matrix:
+        platform: [linux/amd64, linux/arm64]
     steps:
-      - name: Check out the repo
+      - name: Checkout
         uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
 
-      - name: Set up QEMU (multi-arch support)
-        uses: docker/setup-qemu-action@v3
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-        with:
-          install: true
-
-      - name: Log in to Docker Hub
-        uses: docker/login-action@v2
+      - name: Login to Docker
+        uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
 
-      - name: Set tags
+      - name: Set image tag
         id: set-tags
         run: |
           REF_NAME=${GITHUB_REF#refs/tags/}
-          TAGS="${{ inputs.image_name }}:$REF_NAME"
-          if [[ "${{ inputs.add_latest }}" == "true" ]]; then
-            TAGS="$TAGS,${{ inputs.image_name }}:latest"
+          if [ "${{ matrix.platform }}" = "linux/amd64" ]; then
+            IMAGE="${{ inputs.image_name }}:$REF_NAME-amd64"
+          else
+            IMAGE="${{ inputs.image_name }}:$REF_NAME-arm64"
           fi
-          echo "tags=$TAGS" >> $GITHUB_OUTPUT
+          echo "image=$IMAGE" >> $GITHUB_OUTPUT
 
-      - name: Build amd64 image
+      - name: Build and push image
         uses: docker/build-push-action@v6
         with:
           context: .
           file: ${{ inputs.dockerfile }}
-          platforms: linux/amd64
+          platforms: ${{ matrix.platform }}
           push: true
-          tags: ${{ inputs.image_name }}:amd64
+          tags: ${{ steps.set-tags.outputs.image }}
           target: ${{ inputs.target }}
-          cache-from: type=registry,ref=${{ inputs.image_name }}:buildcache-amd64
-          cache-to: type=registry,ref=${{ inputs.image_name }}:buildcache-amd64,mode=max
 
-      - name: Build arm64 image
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          file: ${{ inputs.dockerfile }}
-          platforms: linux/arm64
-          push: true
-          tags: ${{ inputs.image_name }}:arm64
-          target: ${{ inputs.target }}
-          cache-from: type=registry,ref=${{ inputs.image_name }}:buildcache-arm64
-          cache-to: type=registry,ref=${{ inputs.image_name }}:buildcache-arm64,mode=max
-
-      - name: Create and push multi-arch manifest
+      - name: Prepare artifact name
+        id: artifact
         run: |
+          platform=${{ matrix.platform }}
+          echo "safe_platform=${platform//\//-}" >> $GITHUB_OUTPUT
+          echo "${{ steps.set-tags.outputs.image }}" > image.txt
+
+      - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f
+        with:
+          name: image-${{ steps.artifact.outputs.safe_platform }}
+          path: image.txt
+          if-no-files-found: warn
+          compression-level: 6
+          overwrite: false
+          include-hidden-files: false
+
+  create-manifest:
+    name: Create multi-arch Docker manifest
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131
+        with:
+          name: image-linux-amd64
+          path: /tmp/digests/amd64
+
+      - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131
+        with:
+          name: image-linux-arm64
+          path: /tmp/digests/arm64
+
+      - name: Login to Docker
+        uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+
+      - name: Create and push manifest list
+        working-directory: /tmp/digests
+        run: |
+          DOCKER_IMAGES="$(cat amd64/image.txt) $(cat arm64/image.txt)"
           REF_NAME=${GITHUB_REF#refs/tags/}
-          docker manifest create ${{ inputs.image_name }}:$REF_NAME \
-            --amend ${{ inputs.image_name }}:amd64 \
-            --amend ${{ inputs.image_name }}:arm64
-          docker manifest push ${{ inputs.image_name }}:$REF_NAME
-          if [[ "${{ inputs.add_latest }}" == "true" ]]; then
-            docker manifest create ${{ inputs.image_name }}:latest \
-              --amend ${{ inputs.image_name }}:amd64 \
-              --amend ${{ inputs.image_name }}:arm64
-            docker manifest push ${{ inputs.image_name }}:latest
-          fi
+          MANIFEST_IMAGE="${{ inputs.image_name }}:$REF_NAME"
+
+          docker buildx imagetools create $DOCKER_IMAGES -t $MANIFEST_IMAGE
+          docker buildx imagetools inspect $MANIFEST_IMAGE
+
+          if [ "${{ inputs.add_latest }}" = "true" ]; then
+            docker buildx imagetools create $DOCKER_IMAGES -t ${{ inputs.image_name }}:latest
+            docker buildx imagetools inspect ${{ inputs.image_name }}:latest
+          fi

CITATION.cff

@@ -22,5 +22,5 @@ keywords:
   - self-hosted
   - portabase
 license: Apache-2.0
-version: 1.0.3-rc.8
-date-released: "2026-01-28"
+version: 1.1.2-rc.2
+date-released: "2026-02-16"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "portabase-agent"
-version = "1.0.3-rc.8"
+version = "1.1.2-rc.2"
 edition = "2024"
 
 [dependencies]
@@ -17,7 +17,7 @@ base64 = "0.22.1"
 thiserror = "2.0.17"
 log = "0.4.29"
 toml = "0.9.10"
-reqwest = { version = "0.13.1", features = ["json", "blocking", "multipart"] }
+reqwest = { version = "0.13.1", features = ["json", "blocking", "multipart", "stream", "query"] }
 anyhow = "1.0.100"
 tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs"] }
 async-trait = "0.1.89"
@@ -28,11 +28,29 @@ flate2 = "1.1.5"
 tar = "0.4.44"
 tokio-postgres = "0.7.15"
 futures = "0.3.31"
-tracing-log = "0.2.0"
 tracing-appender = "0.2.4"
 time = { version = "0.3.44", features = ["macros"] }
 mongodb = "3.5.0"
+rand = "0.9.2"
+bytes = "1.11.0"
+async-stream = "0.3.6"
+uuid = { version = "1.20.0", features = ["v4"] }
+tokio-util = "0.7.18"
+aws-config = "1.8.13"
+aws-sdk-s3 = { version = "1.122.0", features = ["behavior-version-latest"] }
+async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }
+tokio-tar = "0.3.1"
+oauth2 = "5.0.0"
+hyper = "1.8.1"
+async-http-client = "0.2.0"
+aes-gcm = "0.11.0-rc.3"
+generic-array = "0.14.7"
+futures-util = "0.3.31"
+tokio-stream = "0.1.18"
+aes = "0.9.0-rc.4"
+typenum = "1.19.0"
+
 
 [[bin]]
 name = "app"
-path = "src/main.rs"
+path = "src/main.rs"

Makefile

@@ -15,6 +15,10 @@ seed-mysql:
 	@echo "Seeding MySQL..."
 	mysql -h 127.0.0.1 -P "$$MYSQL_PORT" -u "$$MYSQL_USER" -p"$$MYSQL_PASSWORD" "$$MYSQL_DB" < ./scripts/mysql/seed-mysql.sql
 
+seed-mysql-1gb:
+	@echo "Seeding MySQL..."
+	mysql -h 127.0.0.1 -P "$$MYSQL_PORT" -u "$$MYSQL_USER" -p"$$MYSQL_PASSWORD" "$$MYSQL_DB" < ./scripts/mysql/seed-1gb.sql
+
 
 seed-postgres:
 	@echo "Seeding Postgres..."

docker-compose.prod.yml

@@ -1,10 +1,11 @@
 services:
   rust-app:
-    #    build:
-    #      context: .
-    #      dockerfile: docker/Dockerfile
-    #      target: prod
+#    build:
+#      context: .
+#      dockerfile: docker/Dockerfile
+#      target: prod
     image: portabase/agent:latest
+#    platform: linux/arm64
     container_name: rust-prod
     volumes:
       - ./databases.json:/config/config.json
@@ -13,7 +14,7 @@ services:
       LOG: info
       TZ: "Europe/Paris"
       #      DATABASES_CONFIG_FILE: "config.toml"
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiY2VlZmNmNDQtOGE0YS00NjZlLTkwNDEtN2QzNDMzZjRjOTJkIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUFuYUVKdnVBSExlNGF5d0RmNGplelVobm45VXRkdksyZ3pEMEg2cERJYXczYkJKRkpwVnVDXG5uVFV3MXA3Q2RnOXBzdjZhRnpyOXZPd0J2MjMzckxpdVpCT2lCb2p2Q0QrSlZid3hyTzBRRW5hN2dmaHV1ZGYwXG5VVlJOMkxmK1g1aTkvZzJTNm5xcExoTm1DaGFJNk8ybktYZUNlRmtubEErRUJrNnFoV1FCVGozb05TYTFTOFY1XG40UFRTT2I4NUo3a2k5YllEbXRiNWxrU3dCNXdXOTdtQjg0ZzI2WHAvU3FFcmhKc0NGK3YrN09vTWYzTzJqTTNoXG5XMUQ0MzBPRitWaklwUGdoV09rZy96NXZQUWFHRzhqQ0h4VDlJR0Q0bjhyS05LQ3FTOGNyN2diTGU0cWpNdmhvXG5BQVVvaHpHR2FRNkhlWlJ4S0UvM3J1a2JldnY5dnJ2TTNRSURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiOGZmMDE4NTQtYjJhMS00ZTE0LTkwMjctZTJiOWIxZjQ1YzdlIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUE5TWV4M2pmdnVLdFB5YU1ERnh2Ulp2dmd3YkRJQ2JzQi81Wll5NDNSVVRBaXZRYjJiSDdYXG5qRHBQd1lJeCs4UFBrbHlRbDVMQzV1UWZEaCs4SVd4OG1LZ3FvMXpWMkdiZXdGbEdEWFYxVEdyU1ZEU25aSWR4XG52bWdYc29EeXhVMlJvWUFUMS9YMWxuc2YxenZKdkFMTkhXdEhRdk42SjVDZTFSMmFsendVRGFEVXlJNzRmSldQXG5tNTh0SDMrYklXL0VVTXdjaWNxM0oySWw3Vm9KNkZNUHJQL1ZSOWEvdFF1SU1qa200MXpFY2NscExPa2luRkxuXG54NmVUWkFSZUpya2UrbnRvZ2t4TGEyRWV5a1lUNzB4V3hKNWp5ZExBVnRvNkkyQlVLVVJoTkowTUFaU29NYUtvXG5iMGJRcnY1UzExZWllMnMrT2I3aTYzSFpkVUx0UmV1MVJ3SURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
     extra_hosts:
       - "localhost:host-gateway"
     networks:
@@ -21,6 +22,48 @@ services:
 
 
 
+  db-mongodb-auth:
+    container_name: db-mongodb-auth
+    image: mongo:latest
+    ports:
+      - "27082:27017"
+    environment:
+      MONGO_INITDB_ROOT_USERNAME: root
+      MONGO_INITDB_ROOT_PASSWORD: rootpassword
+      MONGO_INITDB_DATABASE: testdbauth
+    command: mongod --auth
+    networks:
+      - portabase
+    volumes:
+      - mongodb-data-auth:/data/db
+    healthcheck:
+      test: [ "CMD", "mongo", "--eval", "db.adminCommand('ping')" ]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  db-mongodb:
+    container_name: db-mongodb
+    image: mongo:latest
+    ports:
+      - "27083:27017"
+    volumes:
+      - mongodb-data:/data/db
+    healthcheck:
+      test: [ "CMD", "mongosh", "--eval", "db.adminCommand('ping')" ]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+    environment:
+      MONGO_INITDB_DATABASE: testdb
+    networks:
+      - portabase
+
+volumes:
+  mongodb-data:
+  mongodb-data-auth:
+
+
 networks:
   portabase:
     name: portabase_network

docker-compose.yml

@@ -4,7 +4,6 @@ services:
       context: .
       dockerfile: docker/Dockerfile
       target: dev
-    platform: linux/arm64
     container_name: rust-dev
     volumes:
       - .:/app
@@ -12,12 +11,12 @@ services:
       #      - ./databases.toml:/config/config.toml
       - cargo-registry:/usr/local/cargo/registry
       - cargo-git:/usr/local/cargo/git
-      - cargo-target:/app/target
+#      - cargo-target:/app/target
     environment:
       APP_ENV: development
       LOG: debug
       TZ: "Europe/Paris"
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNmExMzc0M2ItYjMwZS00Zjg4LWJjY2EtMmMwZjE2NWVjYjQxIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUE5TWV4M2pmdnVLdFB5YU1ERnh2Ulp2dmd3YkRJQ2JzQi81Wll5NDNSVVRBaXZRYjJiSDdYXG5qRHBQd1lJeCs4UFBrbHlRbDVMQzV1UWZEaCs4SVd4OG1LZ3FvMXpWMkdiZXdGbEdEWFYxVEdyU1ZEU25aSWR4XG52bWdYc29EeXhVMlJvWUFUMS9YMWxuc2YxenZKdkFMTkhXdEhRdk42SjVDZTFSMmFsendVRGFEVXlJNzRmSldQXG5tNTh0SDMrYklXL0VVTXdjaWNxM0oySWw3Vm9KNkZNUHJQL1ZSOWEvdFF1SU1qa200MXpFY2NscExPa2luRkxuXG54NmVUWkFSZUpya2UrbnRvZ2t4TGEyRWV5a1lUNzB4V3hKNWp5ZExBVnRvNkkyQlVLVVJoTkowTUFaU29NYUtvXG5iMGJRcnY1UzExZWllMnMrT2I3aTYzSFpkVUx0UmV1MVJ3SURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiOWZjYjhmNjUtNTcyMC00N2M3LThkNWItYWFlMmVjOTc2OWU1IiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
       #POOLING: 1
       #DATABASES_CONFIG_FILE: "config.toml"
     extra_hosts:
@@ -25,83 +24,83 @@ services:
     networks:
       - portabase
 
-#  db-postgres:
-#    container_name: db-postgres
-#    image: postgres:17-alpine
-#    ports:
-#      - "5436:5432"
-#    volumes:
-#      - postgres-data:/var/lib/postgresql/data
-#    environment:
-#      - POSTGRES_DB=devdb
-#      - POSTGRES_USER=devuser
-#      - POSTGRES_PASSWORD=changeme
-#    networks:
-#      - portabase
+  db-postgres:
+    container_name: db-postgres
+    image: postgres:17-alpine
+    ports:
+      - "5436:5432"
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    environment:
+      - POSTGRES_DB=devdb
+      - POSTGRES_USER=devuser
+      - POSTGRES_PASSWORD=changeme
+    networks:
+      - portabase
+
+  db-mariadb:
+    container_name: db-mariadb
+    image: mariadb:latest
+    ports:
+      - "3311:3306"
+    environment:
+      - MYSQL_DATABASE=mariadb
+      - MYSQL_USER=mariadb
+      - MYSQL_PASSWORD=changeme
+      - MYSQL_RANDOM_ROOT_PASSWORD=yes
+    volumes:
+      - mariadb-data:/var/lib/mysql
+    networks:
+      - portabase
+
 #
-#  db-mariadb:
-#    container_name: db-mariadb
-#    image: mariadb:latest
+#  db-mongodb-auth:
+#    container_name: db-mongodb-auth
+#    image: mongo:latest
 #    ports:
-#      - "3311:3306"
+#      - "27082:27017"
 #    environment:
-#      - MYSQL_DATABASE=mariadb
-#      - MYSQL_USER=mariadb
-#      - MYSQL_PASSWORD=changeme
-#      - MYSQL_RANDOM_ROOT_PASSWORD=yes
-#    volumes:
-#      - mariadb-data:/var/lib/mysql
+#      MONGO_INITDB_ROOT_USERNAME: root
+#      MONGO_INITDB_ROOT_PASSWORD: rootpassword
+#      MONGO_INITDB_DATABASE: testdbauth
+#    command: mongod --auth
+#    networks:
+#      - portabase
+#    volumes:
+#      - mongodb-data-auth:/data/db
+#    healthcheck:
+#      test: [ "CMD", "mongo", "--eval", "db.adminCommand('ping')" ]
+#      interval: 5s
+#      timeout: 5s
+#      retries: 10
+#
+#  db-mongodb:
+#    container_name: db-mongodb
+#    image: mongo:latest
+#    ports:
+#      - "27083:27017"
+#    volumes:
+#      - mongodb-data:/data/db
+#    healthcheck:
+#      test: [ "CMD", "mongosh", "--eval", "db.adminCommand('ping')" ]
+#      interval: 5s
+#      timeout: 5s
+#      retries: 10
+#    environment:
+#      MONGO_INITDB_DATABASE: testdb
 #    networks:
 #      - portabase
-
-
-  db-mongodb-auth:
-    container_name: db-mongodb-auth
-    image: mongo:latest
-    ports:
-      - "27082:27017"
-    environment:
-      MONGO_INITDB_ROOT_USERNAME: root
-      MONGO_INITDB_ROOT_PASSWORD: rootpassword
-      MONGO_INITDB_DATABASE: testdbauth
-    command: mongod --auth
-    networks:
-      - portabase
-    volumes:
-      - mongodb-data-auth:/data/db
-    healthcheck:
-      test: [ "CMD", "mongo", "--eval", "db.adminCommand('ping')" ]
-      interval: 5s
-      timeout: 5s
-      retries: 10
-
-  db-mongodb:
-    container_name: db-mongodb
-    image: mongo:latest
-    ports:
-      - "27083:27017"
-    volumes:
-      - mongodb-data:/data/db
-    healthcheck:
-      test: [ "CMD", "mongosh", "--eval", "db.adminCommand('ping')" ]
-      interval: 5s
-      timeout: 5s
-      retries: 10
-    environment:
-      MONGO_INITDB_DATABASE: testdb
-    networks:
-      - portabase
 
 volumes:
   cargo-registry:
   cargo-git:
-  cargo-target:
+#  cargo-target:
 
 
-#  postgres-data:
-#  mariadb-data:
-  mongodb-data:
-  mongodb-data-auth:
+  postgres-data:
+  mariadb-data:
+#  mongodb-data:
+#  mongodb-data-auth:
 
 networks:
   portabase:

docker/Dockerfile

@@ -1,9 +1,7 @@
 # =========================
-# Base Rust image
+# Base image (shared)
 # =========================
-FROM rust:1.92 AS base
-
-ARG TARGETARCH
+FROM rust:1.92.0 AS base
 
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     pkg-config \
@@ -17,36 +15,44 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     zlib1g \
     curl \
     mariadb-client \
-    wget \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
 
-# =========================
-# PostgreSQL client binaries (only target arch)
-# =========================
-RUN for v in 12 13 14 15 16 17 18; do mkdir -p /usr/lib/postgresql/$v/bin; done
-COPY assets/tools/$TARGETARCH/postgresql/ /tmp/pg/
-RUN for v in 12 13 14 15 16 17 18; do \
-        cp -r /tmp/pg/postgresql-$v/bin/* /usr/lib/postgresql/$v/bin/; \
-    done && rm -rf /tmp/pg && chmod +x /usr/lib/postgresql/*/bin/*
+ARG TARGETARCH
 
 # =========================
-# MongoDB client binaries (only target arch)
+# PostgreSQL client binaries (versions 12-18)
 # =========================
+
+RUN for v in 12 13 14 15 16 17 18; do \
+        mkdir -p /usr/lib/postgresql/$v/bin; \
+    done
+
+COPY assets/tools/amd64/postgresql/ /tmp/pg-x64/
+COPY assets/tools/arm64/postgresql/ /tmp/pg-arm/
+
 RUN if [ "$TARGETARCH" = "amd64" ]; then \
-        wget -q https://fastdl.mongodb.org/tools/db/mongodb-database-tools-debian12-x86_64-100.10.0.deb -O /tmp/mongodb-tools.deb; \
+        for v in 12 13 14 15 16 17 18; do \
+            cp -r /tmp/pg-x64/postgresql-$v/bin/* /usr/lib/postgresql/$v/bin/; \
+        done; \
     elif [ "$TARGETARCH" = "arm64" ]; then \
-        wget -q https://fastdl.mongodb.org/tools/db/mongodb-database-tools-ubuntu2204-arm64-100.10.0.deb -O /tmp/mongodb-tools.deb; \
-    fi && dpkg -i /tmp/mongodb-tools.deb || apt-get install -f -y --no-install-recommends && \
-    rm -f /tmp/mongodb-tools.deb && \
-    mkdir -p /usr/local/mongodb/bin && \
-    ln -sf /usr/bin/mongodump /usr/local/mongodb/bin/mongodump || true && \
-    ln -sf /usr/bin/mongorestore /usr/local/mongodb/bin/mongorestore || true
+        for v in 12 13 14 15 16 17 18; do \
+            cp -r /tmp/pg-arm/postgresql-$v/bin/* /usr/lib/postgresql/$v/bin/; \
+        done; \
+    fi && \
+    rm -rf /tmp/pg-x64 /tmp/pg-arm && \
+    chmod +x /usr/lib/postgresql/*/bin/*
+
+# =========================
+# MongoDB client binaries
+# =========================
+COPY assets/tools/${TARGETARCH}/mongodb/ /usr/local/mongodb/
+RUN chmod +x /usr/local/mongodb/bin/*
 
 WORKDIR /app
 
 # =========================
-# Development stage
+# Development image
 # =========================
 FROM base AS dev
 
@@ -57,26 +63,15 @@ RUN mkdir src && echo "fn main() {}" > src/main.rs
 RUN cargo build
 RUN rm -rf src
 
-COPY . .
-
 COPY entrypoint.sh /entrypoint.sh
 RUN chmod +x /entrypoint.sh
 
 CMD ["/entrypoint.sh"]
 
 # =========================
-# Builder stage for production
+# Builder (production)
 # =========================
 FROM base AS builder
-WORKDIR /app
-
-RUN cargo install cargo-chef
-
-COPY Cargo.toml Cargo.lock ./
-RUN mkdir src && echo "fn main() {}" > src/main.rs
-RUN cargo chef prepare --recipe-path recipe.json
-RUN cargo chef cook --recipe-path recipe.json --release
-RUN rm -rf src
 
 COPY . .
 RUN cargo build --release
@@ -84,7 +79,7 @@ RUN cargo build --release
 RUN echo "APP_VERSION=$(cargo pkgid | awk -F# '{print $2}')" > /app/version.env
 
 # =========================
-# Production runtime
+# Runtime (production)
 # =========================
 FROM ubuntu:24.04 AS prod
 
@@ -99,16 +94,17 @@ RUN apt-get update && apt-get install -y \
     mariadb-client \
     && rm -rf /var/lib/apt/lists/*
 
+
 WORKDIR /app
 
 COPY --from=builder /app/target/release/app /usr/local/bin/app
 COPY --from=builder /app/version.env /app/version.env
-
 COPY entrypoint.sh /entrypoint.sh
 RUN chmod +x /entrypoint.sh
 
 COPY --from=base /usr/lib/postgresql/ /usr/lib/postgresql/
-COPY --from=base /usr/local/mongodb/bin /usr/local/mongodb/bin
+COPY --from=base /usr/local/mongodb/bin/ /usr/local/mongodb/bin/
+
 
 ENV APP_ENV=production
 

scripts/README.md

@@ -1,11 +1,12 @@
 # Seed instructions
 
-## MongoDB 
+## MongoDB
 
 ```bash
 make seed-mongo     
 make seed-mongo-auth  
 make seed-mysql      
+make seed-mysql-1gb    
 make seed-postgres      
 make seed-postgres-1gb  
 make seed-all      

scripts/mysql/seed-1gb.sql

@@ -0,0 +1,53 @@
+CREATE DATABASE IF NOT EXISTS mariadb;
+USE mariadb;
+
+DROP TABLE IF EXISTS users;
+DROP TABLE IF EXISTS products;
+
+CREATE TABLE users (
+                       id BIGINT AUTO_INCREMENT PRIMARY KEY,
+                       username VARCHAR(50) NOT NULL UNIQUE,
+                       email VARCHAR(100) NOT NULL UNIQUE,
+                       password VARCHAR(255) NOT NULL,
+                       created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+) ENGINE=InnoDB;
+
+CREATE TABLE products (
+                          id BIGINT AUTO_INCREMENT PRIMARY KEY,
+                          name VARCHAR(100) NOT NULL,
+                          description MEDIUMTEXT,
+                          price DECIMAL(10,2) NOT NULL,
+                          created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+) ENGINE=InnoDB;
+
+
+DELIMITER $$
+
+DROP PROCEDURE IF EXISTS seed_data$$
+CREATE PROCEDURE seed_data()
+BEGIN
+    DECLARE i BIGINT DEFAULT 1;
+    DECLARE j BIGINT;
+    DECLARE large_text TEXT;
+
+    SET large_text = REPEAT('Lorem ipsum dolor sit amet, consectetur adipiscing elit. ', 300);
+
+    WHILE i <= 200000 DO
+        INSERT INTO users (username, email, password)
+        VALUES (CONCAT('user', i), CONCAT('user', i, '@example.com'), 'changeme');
+        SET i = i + 1;
+END WHILE;
+
+    SET i = 1;
+    WHILE i <= 200000 DO
+        INSERT INTO products (name, description, price)
+        VALUES (CONCAT('Product ', i), large_text, ROUND(RAND()*1000,2));
+        SET i = i + 1;
+END WHILE;
+END$$
+
+DELIMITER ;
+
+CALL seed_data();
+
+DROP PROCEDURE IF EXISTS seed_data;

src/core/agent.rs

@@ -52,7 +52,7 @@ impl Agent {
             if db.data.backup.action {
                 let _ = self
                     .backup_service
-                    .dispatch(&db.generated_id, &config, method.clone())
+                    .dispatch(&db.generated_id, &config, method.clone(), &db.storages, db.encrypt)
                     .await;
             } else if db.data.restore.action {
                 let _ = self

src/core/context.rs

@@ -1,3 +1,4 @@
+use crate::services::api::ApiClient;
 use crate::settings::CONFIG;
 use crate::utils::edge_key::{EdgeKey, EdgeKeyError, decode_edge_key};
 use tracing::{debug, error, info};
@@ -6,6 +7,7 @@ use tracing::{debug, error, info};
 pub struct Context {
     #[allow(dead_code)]
     pub edge_key: EdgeKey,
+    pub api: ApiClient,
 }
 
 impl Context {
@@ -33,7 +35,13 @@ impl Context {
                 panic!("Cannot initialize AgentContext due to invalid EDGE_KEY");
             }
         };
+        
+        let server_url = format!("{}/api", edge_key.server_url);
+        let api_client = ApiClient::new(server_url);
 
-        Context { edge_key }
+        Context {
+            edge_key: edge_key,
+            api: api_client,
+        }
     }
 }

src/domain/mysql/ping.rs

@@ -1,19 +1,26 @@
 use crate::services::config::DatabaseConfig;
-use anyhow::Context;
 use std::collections::HashMap;
-use std::process::Command;
+use tokio::process::Command;
+use tokio::time::{Duration, timeout};
 
 pub async fn run(cfg: DatabaseConfig, env: HashMap<String, String>) -> anyhow::Result<bool> {
-    let output = Command::new("mysqladmin")
-        .arg("--host")
+    let mut cmd = Command::new("mysqladmin");
+    cmd.arg("--host")
         .arg(cfg.host)
         .arg("--port")
         .arg(cfg.port.to_string())
         .arg("--user")
         .arg(cfg.username)
         .arg("ping")
-        .envs(env)
-        .output()
-        .with_context(|| format!("Failed to ping MySQL server {}", cfg.name))?;
-    Ok(output.status.success())
+        .envs(env);
+
+    let result = timeout(Duration::from_secs(10), cmd.output()).await;
+
+    match result {
+        Ok(output) => {
+            let output = output?;
+            Ok(output.status.success())
+        }
+        Err(_) => Ok(false),
+    }
 }

src/domain/postgres/database.rs

@@ -28,7 +28,6 @@ impl Database for PostgresDatabase {
         match self.format {
             PostgresDumpFormat::Fc => ".dump",
             PostgresDumpFormat::Fd => ".gz",
-            // PostgresDumpFormat::Fd => ".tar.gz",
         }
     }
 

src/domain/postgres/restore.rs

@@ -86,9 +86,13 @@ pub async fn run(
                     }
                 };
 
+                info!("tar_gz {:?}", tar_gz);
+
                 let dec = flate2::read::GzDecoder::new(tar_gz);
                 let mut archive = tar::Archive::new(dec);
 
+
+
                 let tmp_dir = match tempfile::TempDir::new() {
                     Ok(d) => d,
                     Err(e) => {
@@ -100,12 +104,16 @@ pub async fn run(
                     }
                 };
 
+
+
                 if let Err(e) = archive.unpack(tmp_dir.path()) {
                     error!("Failed to unpack FD archive for {}: {:?}", cfg.name, e);
                     return Err(e.into());
                 }
 
                 debug!("Listing contents of temp dir: {}", tmp_dir.path().display());
+
+
                 for entry in std::fs::read_dir(tmp_dir.path())? {
                     if let Ok(entry) = entry {
                         let path = entry.path();

src/services/api/client.rs

@@ -0,0 +1,83 @@
+#![allow(dead_code)]
+
+use reqwest::{Client, Method};
+use serde::de::DeserializeOwned;
+use std::time::Duration;
+use crate::services::api::ApiError;
+
+#[derive(Clone, Debug)]
+pub struct ApiClient {
+    base_url: String,
+    http: Client,
+}
+
+impl ApiClient {
+    pub fn new(base_url: impl Into<String>) -> Self {
+        let http = Client::builder()
+            .timeout(Duration::from_secs(30))
+            .build()
+            .expect("failed to build http client");
+
+        Self {
+            base_url: base_url.into(),
+            http,
+        }
+    }
+
+    pub async fn request<T: DeserializeOwned>(
+        &self,
+        method: Method,
+        path: &str,
+    ) -> Result<Option<T>, ApiError> {
+        let url = format!("{}{}", self.base_url, path);
+
+        let res = self.http.request(method, &url).send().await?;
+        let status = res.status();
+        let body = res.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            return Err(ApiError::HttpResponse { status, body });
+        }
+
+        if body.trim().is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(serde_json::from_str::<T>(&body)?))
+        }
+    }
+    pub async fn request_with_body<T, B>(
+        &self,
+        method: Method,
+        path: &str,
+        body: &B,
+    ) -> Result<Option<T>, ApiError>
+    where
+        T: DeserializeOwned,
+        B: serde::Serialize,
+    {
+        let url = format!("{}{}", self.base_url, path);
+
+        let res = self
+            .http
+            .request(method, &url)
+            .json(body)
+            .send()
+            .await?;
+
+        let status = res.status();
+        let body_text = res.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            return Err(ApiError::HttpResponse {
+                status,
+                body: body_text,
+            });
+        }
+
+        if body_text.trim().is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(serde_json::from_str::<T>(&body_text)?))
+        }
+    }
+}

src/services/api/endpoints/agent/backup/mod.rs

@@ -0,0 +1,67 @@
+pub mod upload;
+
+use crate::services::api::models::agent::backup::BackupResponse;
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct BackupCreateRequest {
+    pub method: String,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+}
+
+#[derive(Serialize)]
+pub struct BackupUpdateRequest {
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+    pub status: String,
+    pub size: Option<u64>,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+}
+
+
+impl ApiClient {
+    pub async fn backup_create(
+        &self,
+        method: impl Into<String>,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+    ) -> Result<Option<BackupResponse>, ApiError> {
+        let body = BackupCreateRequest {
+            method: method.into(),
+            generated_id: generated_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body)
+            .await
+    }
+
+    pub async fn backup_update(
+        &self,
+        agent_id: impl Into<String>,
+        backup_id: impl Into<String>,
+        status: impl Into<String>,
+        file_size: impl Into<Option<u64>>,
+        generated_id: impl Into<String>,
+    ) -> Result<Option<BackupResponse>, ApiError> {
+        let body = BackupUpdateRequest {
+            backup_id: backup_id.into(),
+            status: status.into(),
+            size: file_size.into(),
+            generated_id: generated_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup", agent_id);
+
+        self.request_with_body(Method::PATCH, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/backup/upload/init.rs

@@ -0,0 +1,37 @@
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+use crate::services::api::models::agent::backup::BackupUploadResponse;
+
+#[derive(Serialize)]
+pub struct InitUploadRequest {
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    #[serde(rename = "storageChannelId")]
+    pub storage_channel_id: String,
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+}
+
+impl ApiClient {
+    pub async fn backup_upload_init(
+        &self,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+        storage_channel_id: impl Into<String>,
+        backup_id: impl Into<String>,
+    ) -> Result<Option<BackupUploadResponse>, ApiError> {
+        let body = InitUploadRequest {
+            generated_id: generated_id.into(),
+            storage_channel_id: storage_channel_id.into(),
+            backup_id: backup_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup/upload/init", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/backup/upload/mod.rs

@@ -0,0 +1,2 @@
+pub mod init;
+pub mod status;

src/services/api/endpoints/agent/backup/upload/status.rs

@@ -0,0 +1,46 @@
+use crate::services::api::models::agent::backup::BackupUploadResponse;
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct StatusUploadRequest {
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    #[serde(rename = "backupStorageId")]
+    pub backup_storage_id: String,
+    pub status: String,
+    pub path: String,
+    pub size: u64,
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+}
+
+impl ApiClient {
+    pub async fn backup_upload_status(
+        &self,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+        backup_storage_id: impl Into<String>,
+        status: impl Into<String>,
+        remote_path: impl Into<String>,
+        total_size: impl Into<u64>,
+        backup_id: impl Into<String>,
+    ) -> Result<Option<BackupUploadResponse>, ApiError> {
+        let body = StatusUploadRequest {
+            generated_id: generated_id.into(),
+            backup_storage_id: backup_storage_id.into(),
+            status: status.into(),
+            path: remote_path.into(),
+            size: total_size.into(),
+            backup_id: backup_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup/upload/status", agent_id);
+
+        self.request_with_body(Method::PATCH, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/mod.rs

@@ -0,0 +1,2 @@
+pub mod status;
+pub mod backup;

src/services/api/endpoints/agent/status.rs

@@ -0,0 +1,36 @@
+use crate::services::api::models::agent;
+use crate::services::api::{ApiClient, ApiError};
+use agent::status::PingResult;
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct DatabasePayload<'a> {
+    pub name: &'a str,
+    pub dbms: &'a str,
+    #[serde(rename = "generatedId")]
+    pub generated_id: &'a str,
+}
+
+#[derive(Serialize)]
+pub struct StatusRequest<'a> {
+    pub version: &'a str,
+    pub databases: Vec<DatabasePayload<'a>>,
+}
+
+impl ApiClient {
+    pub async fn agent_status<'a>(
+        &self,
+        agent_id: impl Into<String>,
+        version: &'a str,
+        databases: Vec<DatabasePayload<'a>>,
+    ) -> Result<Option<PingResult>, ApiError> {
+        let body = StatusRequest { version, databases };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/status", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body).await
+    }
+}

src/services/api/endpoints/mod.rs

@@ -0,0 +1,3 @@
+pub mod agent;
+
+pub use agent::status;

src/services/api/error.rs

@@ -0,0 +1,21 @@
+use reqwest::StatusCode;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+#[allow(dead_code)]
+pub enum ApiError {
+    #[error("http client error: {0}")]
+    Http(#[from] reqwest::Error),
+
+    #[error("serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("api error: status={status}, body={body}")]
+    HttpResponse {
+        status: StatusCode,
+        body: String,
+    },
+
+    #[error("api returned unexpected response")]
+    UnexpectedResponse,
+}

src/services/api/mod.rs

@@ -0,0 +1,8 @@
+pub mod client;
+pub mod error;
+pub mod models;
+pub mod endpoints;
+
+pub use client::ApiClient;
+pub use error::ApiError;
+

src/services/api/models/agent/backup.rs

@@ -0,0 +1,24 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct BackupStorage {
+    pub id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BackupUploadResponse {
+    pub message: String,
+    #[serde(rename = "backupStorage")]
+    pub backup_storage: BackupStorage,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct Backup {
+    pub id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BackupResponse {
+    pub message: String,
+    pub backup: Backup,
+}

src/services/api/models/agent/mod.rs

@@ -0,0 +1,2 @@
+pub mod status;
+pub mod backup;

src/services/api/models/agent/status.rs

@@ -0,0 +1,57 @@
+#![allow(dead_code)]
+
+use serde::{Deserialize, Serialize};
+use toml::Value;
+use crate::utils::deserializer::deserialize_snake_case;
+
+#[derive(Debug, Deserialize)]
+pub struct PingResult {
+    pub agent: AgentInfo,
+    pub databases: Vec<DatabaseStatus>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct AgentInfo {
+    pub id: String,
+    #[serde(rename = "lastContact")]
+    pub last_contact: String,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+#[serde(rename_all = "snake_case")]
+pub struct DatabaseStorage {
+    pub id: String,
+    #[serde(deserialize_with = "deserialize_snake_case")]
+    pub config: Value,
+    pub provider: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DatabaseStatus {
+    pub dbms: String,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    pub storages: Vec<DatabaseStorage>,
+    pub encrypt: bool,
+    pub data: DatabaseData,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DatabaseData {
+    pub backup: BackupInfo,
+    pub restore: RestoreInfo,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct BackupInfo {
+    pub action: bool,
+    pub cron: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RestoreInfo {
+    pub action: bool,
+    pub file: Option<String>,
+    #[serde(rename = "metaFile")]
+    pub meta_file: Option<String>,
+}

src/services/api/models/mod.rs

@@ -0,0 +1,2 @@
+pub mod agent;
+

src/services/backup.rs

@@ -1,26 +1,21 @@
 #![allow(dead_code)]
 
-use crate::core::context::Context;
+use crate::core::context::Context as CoreContext;
 use crate::domain::factory::DatabaseFactory;
+use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::config::{DatabaseConfig, DatabasesConfig, DbType};
+use crate::services::storage;
 use crate::utils::common::BackupMethod;
-use crate::utils::file::full_extension;
+use crate::utils::compress::compress_to_tar_gz_large;
 use anyhow::Result;
-use hex;
-use openssl::encrypt::Encrypter;
-use openssl::hash::MessageDigest;
-use openssl::pkey::PKey;
-use openssl::rand::rand_bytes;
-use openssl::rsa::Padding;
-use openssl::symm::{Cipher, Crypter, Mode};
-use reqwest::multipart::{Form, Part};
+use futures::future::join_all;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use tempfile::TempDir;
-use tokio::fs;
 use tracing::{error, info};
+use crate::utils::locks::FileLock;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BackupResult {
     pub generated_id: String,
     pub db_type: DbType,
@@ -29,12 +24,21 @@ pub struct BackupResult {
     pub code: Option<String>,
 }
 
+#[derive(Debug)]
+pub struct UploadResult {
+    pub storage_id: String,
+    pub success: bool,
+    pub error: Option<String>,
+    pub remote_file_path: Option<String>,
+    pub total_size: Option<u64>,
+}
+
 pub struct BackupService {
-    ctx: Arc<Context>,
+    ctx: Arc<CoreContext>,
 }
 
 impl BackupService {
-    pub fn new(ctx: Arc<Context>) -> Self {
+    pub fn new(ctx: Arc<CoreContext>) -> Self {
         Self { ctx }
     }
 
@@ -43,6 +47,8 @@ impl BackupService {
         generated_id: &String,
         config: &DatabasesConfig,
         method: BackupMethod,
+        storages: &Vec<DatabaseStorage>,
+        encrypt: bool,
     ) {
         if let Some(cfg) = config
             .databases
@@ -50,22 +56,113 @@ impl BackupService {
             .find(|c| c.generated_id == generated_id.as_str())
         {
             let db_cfg = cfg.clone();
-            let ctx_clone = self.ctx.clone();
+            let ctx = self.ctx.clone();
+            let storages_clone = storages.clone();
+            let generated_id_clone = generated_id.clone();
 
             tokio::spawn(async move {
                 match TempDir::new() {
                     Ok(temp_dir) => {
-                        let tmp_path = temp_dir.path().to_path_buf();
-                        info!("Created temp directory {}", tmp_path.display());
-
-                        match BackupService::run(db_cfg, &tmp_path).await {
-                            Ok(result) => {
-                                let service = BackupService { ctx: ctx_clone };
-                                service.send_result(result, method).await;
+                        match FileLock::is_locked(&generated_id_clone).await {
+                            Ok(true) => {
+                                error!("Backup already running for {}", &generated_id_clone);
+                                return;
                             }
-                            Err(e) => error!("Backup error {}", e),
+                            Ok(false) => {
+                                match ctx
+                                    .api
+                                    .backup_create(
+                                        method.clone().to_string(),
+                                        ctx.edge_key.agent_id.clone(),
+                                        &generated_id_clone,
+                                    )
+                                    .await
+                                {
+                                    Ok(backup_created_result) => {
+                                        info!("Backup created successfully");
+                                        let tmp_path = temp_dir.path().to_path_buf();
+                                        info!("Created temp directory {}", tmp_path.display());
+                                        match BackupService::run(db_cfg, &tmp_path).await {
+                                            Ok(mut result) => {
+                                                let backup_id = backup_created_result.unwrap().backup.id;
+
+                                                if result.status == "failed" {
+                                                    error!("Backup failed early for {}", result.generated_id);
+
+                                                    let service = BackupService { ctx: ctx.clone() };
+
+                                                    let _ = service
+                                                        .send_result(result, vec![], &backup_id)
+                                                        .await;
+
+                                                    return;
+                                                }
+
+
+                                                if let Some(backup_file) = result.backup_file.take() {
+                                                    match compress_to_tar_gz_large(&backup_file).await {
+                                                        Ok(compression_result) => {
+                                                            result.backup_file =
+                                                                Some(compression_result.compressed_path);
+                                                            let service = BackupService { ctx: ctx.clone() };
+                                                            match service
+                                                                .upload(
+                                                                    result.clone(),
+                                                                    method,
+                                                                    storages_clone.clone(),
+                                                                    encrypt,
+                                                                    &backup_id,
+                                                                )
+                                                                .await
+                                                            {
+                                                                Ok(upload_result) => {
+                                                                    match service
+                                                                        .send_result(
+                                                                            result,
+                                                                            upload_result,
+                                                                            &backup_id,
+                                                                        )
+                                                                        .await
+                                                                    {
+                                                                        Ok(_) => {
+                                                                            return;
+                                                                        }
+                                                                        Err(e) => {
+                                                                            error!(
+                                                                        "Failed to send backup result: {}",
+                                                                        e
+                                                                    );
+                                                                        }
+                                                                    }
+                                                                }
+                                                                Err(e) => {
+                                                                    error!(
+                                                                "Failed to upload backup files: {}",
+                                                                e
+                                                            );
+                                                                }
+                                                            }
+                                                        }
+                                                        Err(e) => {
+                                                            error!(
+                                                        "Failed to compress backup file : {}",
+                                                        e
+                                                    );
+                                                        }
+                                                    }
+                                                } else {
+                                                    error!("No backup file generated");
+                                                }
+                                            }
+                                            Err(e) => error!("BackupService run failed: {}", e),
+                                        }
+                                        // TempDir is automatically deleted when dropped here
+                                    }
+                                    Err(e) => error!("Backup creation failed: {}", e),
+                                }
+                            }
+                            Err(e) => error!("An error occurred while checking lock : {}", e),
                         }
-                        // TempDir is automatically deleted when dropped here
                     }
                     Err(e) => error!("Failed to create temp dir: {}", e),
                 }
@@ -78,7 +175,16 @@ impl BackupService {
         let generated_id = cfg.generated_id.clone();
         let db_type = cfg.db_type.clone();
 
-        let reachable = db_instance.ping().await.unwrap_or(false);
+
+        let reachable = match db_instance.ping().await {
+            Ok(v) => v,
+            Err(e) => {
+                error!("Ping failed: {}", e);
+                return Err(e.into());
+            }
+        };
+
+
         info!("Reachable: {}", reachable);
         if !reachable {
             return Ok(BackupResult {
@@ -117,102 +223,191 @@ impl BackupService {
         }
     }
 
-    pub async fn send_result(&self, result: BackupResult, method: BackupMethod) {
+    pub async fn upload(
+        &self,
+        result: BackupResult,
+        method: BackupMethod,
+        storages: Vec<DatabaseStorage>,
+        encrypt: bool,
+        backup_id: &String,
+    ) -> Result<Vec<UploadResult>> {
         if result.code.as_deref() == Some("backup_already_in_progress") {
+            info!("Skipping send: backup already in progress");
+            anyhow::bail!("backup_already_in_progres");
+        }
+
+        let upload_futures = storages.into_iter().map(|storage| {
             info!(
-                "[BackupService] Skipping send for DB {}: backup already in progress",
-                result.generated_id
+                "Uploading storage -> {:?} for {:?}",
+                storage.provider, storage.id
             );
-            return;
-        }
+            let provider = storage::get_provider(&storage);
+            let result_clone = result.clone();
+            let ctx_clone = self.ctx.clone();
+            let storages_clone = storage.clone();
+            let storage_id = storages_clone.id;
+            let generated_id = result_clone.generated_id.clone();
 
-        info!(
-            "[BackupService] DB: {} Type: {} Status: {} File: {:?}",
-            result.generated_id,
-            result.db_type.as_str(),
-            result.status,
-            result.backup_file
-        );
+            async move {
+                match self
+                    .ctx
+                    .api
+                    .backup_upload_init(
+                        self.ctx.edge_key.agent_id.clone(),
+                        generated_id.clone(),
+                        storage_id.clone(),
+                        backup_id,
+                    )
+                    .await
+                {
+                    Ok(upload_init_result) => {
+                        info!("Uploading init result: {:#?}", upload_init_result);
+                        let backup_storage_id = upload_init_result.unwrap().backup_storage.id.clone();
+                        match provider {
+                            Some(provider) => {
+                                let upload_result = provider
+                                    .upload(
+                                        ctx_clone,
+                                        result_clone,
+                                        method,
+                                        &storage,
+                                        Some(encrypt),
+                                    )
+                                    .await;
 
-        let client = reqwest::Client::new();
-        let url = format!(
-            "{}/api/agent/{}/backup",
-            self.ctx.edge_key.server_url, self.ctx.edge_key.agent_id
-        );
+                                let status = if upload_result.success {
+                                    "success"
+                                } else {
+                                    "failed"
+                                };
+                                info!("Storage {} uploaded to remote path {:?}", storage_id, upload_result.remote_file_path);
 
-        let mut form = Form::new()
-            .text("generatedId", result.generated_id.clone())
-            .text("status", result.status.clone())
-            .text("method", method.to_string());
 
-        if let Some(file_path) = result.backup_file {
-            match fs::read(&file_path).await {
-                Ok(raw_data) => {
-                    // AES key + IV
-                    let mut aes_key = [0u8; 32];
-                    rand_bytes(&mut aes_key).unwrap();
+                                let (remote_path, total_size) = match (
+                                    &upload_result.remote_file_path,
+                                    upload_result.total_size,
+                                ) {
+                                    (Some(path), Some(size)) => (path.clone(), size),
+                                    _ => {
+                                        return UploadResult {
+                                            storage_id: storage_id.clone(),
+                                            success: false,
+                                            error: Some("remote_file_path or total_size missing".to_string()),
+                                            remote_file_path: None,
+                                            total_size: None,
+                                        }
+                                    }
+                                };
 
-                    let mut iv = [0u8; 16];
-                    rand_bytes(&mut iv).unwrap();
-
-                    // AES CBC PKCS7 encryption
-                    let cipher = Cipher::aes_256_cbc();
-                    let mut encrypter =
-                        Crypter::new(cipher, Mode::Encrypt, &aes_key, Some(&iv)).unwrap();
-                    encrypter.pad(true);
-                    let mut encrypted = vec![0u8; raw_data.len() + cipher.block_size()];
-                    let count = encrypter.update(&raw_data, &mut encrypted).unwrap();
-                    let rest = encrypter.finalize(&mut encrypted[count..]).unwrap();
-                    encrypted.truncate(count + rest);
-
-                    // Encrypt AES key with RSA public key
-                    let pub_key_pem = self.ctx.edge_key.public_key.as_bytes();
-                    let pkey = PKey::public_key_from_pem(pub_key_pem).unwrap();
-
-                    let mut encrypter = Encrypter::new(&pkey).unwrap();
-                    // Set OAEP padding (default OAEP uses SHA1, so override)
-                    encrypter.set_rsa_padding(Padding::PKCS1_OAEP).unwrap();
-                    // Set OAEP hash to SHA‑256
-                    encrypter.set_rsa_oaep_md(MessageDigest::sha256()).unwrap();
-                    encrypter.set_rsa_mgf1_md(MessageDigest::sha256()).unwrap();
-
-                    let mut encrypted_key = vec![0u8; encrypter.encrypt_len(&aes_key).unwrap()];
-                    let encrypted_len = encrypter.encrypt(&aes_key, &mut encrypted_key).unwrap();
-                    encrypted_key.truncate(encrypted_len);
-
-                    let extension = full_extension(&file_path);
-
-                    // Attach file and AES info to multipart form
-                    form = form
-                        .part(
-                            "file",
-                            Part::bytes(encrypted)
-                                .file_name(format!("{}.enc", result.generated_id)),
-                        )
-                        .text("aes_key", hex::encode(encrypted_key))
-                        .text("iv", hex::encode(iv))
-                        .text("extension", extension);
-                }
-                Err(e) => {
-                    error!("Failed to read backup file: {}", e);
+                                match self.ctx.api.backup_upload_status(
+                                    self.ctx.edge_key.agent_id.clone(),
+                                    generated_id.clone(),
+                                    backup_storage_id,
+                                    status,
+                                    remote_path,
+                                    total_size,
+                                    backup_id,
+                                ).await {
+                                    Ok(_) => {
+                                        upload_result
+                                    }
+                                    Err(err) => {
+                                        error!(
+                                        "backup_upload_status failed (generated_id={}, storage_id={}): {}",
+                                        generated_id, storage_id, err
+                                    );
+                                        UploadResult {
+                                            storage_id: storage_id.clone(),
+                                            success: false,
+                                            error: Some(err.to_string()),
+                                            remote_file_path: None,
+                                            total_size: None,
+                                        }
+                                    }
+                                }
+                            }
+                            None => {
+                                error!("Skipping storage due to missing provider");
+                                UploadResult {
+                                    storage_id: storage_id.clone(),
+                                    success: false,
+                                    error: Some(
+                                        "Skipping storage due to missing provider".to_string(),
+                                    ),
+                                    remote_file_path: None,
+                                    total_size: None,
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        error!(
+                            "Unable to create the storage backup on remote server : {}",
+                            e
+                        );
+                        UploadResult {
+                            storage_id: storage_id.clone(),
+                            success: false,
+                            error: Some(
+                                "Unable to create the storage backup on remote server".to_string(),
+                            ),
+                            remote_file_path: None,
+                            total_size: None,
+                        }
+                    }
                 }
             }
+        });
+
+        let results: Vec<UploadResult> = join_all(upload_futures).await;
+        info!("Upload results: {:#?}", results);
+
+        Ok(results)
+    }
+
+    pub async fn send_result(
+        &self,
+        result: BackupResult,
+        upload_results: Vec<UploadResult>,
+        backup_id: &String,
+    ) -> Result<()> {
+        let status = if upload_results.iter().any(|r| r.success) {
+            "success"
         } else {
-            form = form.text("file", "");
-        }
+            "failed"
+        };
 
-        match client.post(&url).multipart(form).send().await {
-            Ok(resp) => {
-                let status = resp.status();
-                if status.is_success() {
-                    info!("Backup result sent successfully");
-                } else {
-                    let text = resp.text().await.unwrap_or_default(); // consumes resp
-                    error!("Backup result failed, status: {}, body: {}", status, text);
-                }
+        let file_size = if status == "failed" {
+            None
+        } else {
+            let mut sum = 0u64;
+            let mut count = 0u64;
+
+            for size in upload_results.iter().filter_map(|r| r.total_size) {
+                sum += size;
+                count += 1;
             }
+
+            if count == 0 {
+                None
+            } else {
+                Some(sum / count)
+            }
+        };
+
+        match self
+            .ctx
+            .api
+            .backup_update(self.ctx.edge_key.agent_id.clone(), backup_id, status, file_size, &result.generated_id)
+            .await
+        {
+            Ok(_result) => Ok(()),
             Err(e) => {
-                error!("Failed to send backup result: {}", e);
+                error!(
+                    "backup_update failed (generated_id={}, backup_id={}): {}",
+                    &result.generated_id, &backup_id, e
+                );
+                Err(e.into())
             }
         }
     }

src/services/cron.rs

@@ -1,11 +1,13 @@
 #![allow(dead_code)]
 
 use crate::core::context::Context;
-use crate::services::status::DatabaseStatus;
+use crate::utils::common::vec_to_option_json;
 use crate::utils::redis_client;
 use crate::utils::task_manager::cron::check_and_update_cron;
-use std::sync::Arc;
 use redis::aio::MultiplexedConnection;
+use serde_json::{Value, json};
+use std::sync::Arc;
+use crate::services::api::models::agent::status::DatabaseStatus;
 
 pub struct CronService {
     ctx: Arc<Context>,
@@ -23,6 +25,12 @@ impl CronService {
         let dbms = database.dbms.as_str();
         let task_name = format!("periodic.backup_{}", generated_id);
         let args = vec![generated_id.to_string(), dbms.to_string()];
+        let storages: Option<Value> = vec_to_option_json(database.storages.clone());
+        let encrypt: bool = database.encrypt;
+        let metadata = json!({
+            "storages": storages,
+            "encrypt": encrypt
+        });
 
         check_and_update_cron(
             &mut self.conn,
@@ -30,7 +38,9 @@ impl CronService {
             args,
             "tasks.database.periodic_backup",
             task_name,
-        ).await;
+            Option::from(metadata),
+        )
+        .await;
 
         Ok(true)
     }

src/services/mod.rs

@@ -1,5 +1,7 @@
 pub mod config;
-pub mod status;
 pub mod cron;
 pub mod backup;
-pub mod restore;
+pub mod restore;
+mod storage;
+pub mod api;
+pub mod status;

src/services/restore.rs

@@ -1,15 +1,18 @@
 #![allow(dead_code)]
-
+#![warn(unused_assignments)]
 use crate::core::context::Context;
 use crate::domain::factory::DatabaseFactory;
+use crate::services::api::models::agent::status::DatabaseStatus;
 use crate::services::config::{DatabaseConfig, DatabasesConfig};
-use crate::services::status::DatabaseStatus;
+use crate::utils::compress::decompress_large_tar_gz;
+use crate::utils::file::decrypt_file_stream_gcm;
 use anyhow::Result;
-use tracing::{error, info};
+use reqwest::{Client, Url};
 use serde::Serialize;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use tempfile::TempDir;
+use tracing::{error, info};
 
 #[derive(Debug, Serialize)]
 pub struct RestoreResult {
@@ -36,21 +39,30 @@ impl RestoreService {
             let db_cfg = cfg.clone();
             let ctx_clone = self.ctx.clone();
             let file_to_restore = db.data.restore.file.clone();
-
+            if file_to_restore.is_none() {
+                error!("restore file not found");
+                return;
+            }
             tokio::spawn(async move {
                 match TempDir::new() {
                     Ok(temp_dir) => {
                         let tmp_path = temp_dir.path().to_path_buf();
                         info!("Created temp directory {}", tmp_path.display());
-
-                        match RestoreService::run(db_cfg, &tmp_path, &file_to_restore).await {
+                        match RestoreService::run(
+                            &ctx_clone,
+                            db_cfg,
+                            &tmp_path,
+                            &file_to_restore.unwrap(),
+                        )
+                        .await
+                        {
                             Ok(result) => {
                                 let service = RestoreService { ctx: ctx_clone };
                                 service.send_result(result).await;
                             }
                             Err(e) => error!("Restoration error {}", e),
                         }
-                        // TempDir is automatically deleted when dropped here
+                        // TempDir is automatically deleted when dropped
                     }
                     Err(e) => error!("Failed to create temp dir: {}", e),
                 }
@@ -59,6 +71,7 @@ impl RestoreService {
     }
 
     pub async fn run(
+        ctx: &Arc<Context>,
         cfg: DatabaseConfig,
         tmp_path: &Path,
         file_url: &str,
@@ -67,8 +80,9 @@ impl RestoreService {
 
         info!("File url: {}", file_url);
 
-        let client = reqwest::Client::new();
+        let client = Client::new();
         let response = client.get(file_url).send().await?;
+
         if !response.status().is_success() {
             error!("Backup download failed with status {}", response.status());
             return Ok(RestoreResult {
@@ -77,27 +91,96 @@ impl RestoreService {
             });
         }
 
+        let filename_from_header = response
+            .headers()
+            .get(reqwest::header::CONTENT_DISPOSITION)
+            .and_then(|v| v.to_str().ok())
+            .and_then(|s| s.split("filename=").nth(1))
+            .map(|f| f.trim_matches('"').to_string());
+
+        let filename_from_url = Url::parse(file_url).ok().and_then(|u| {
+            u.path_segments()?
+                .last()
+                .filter(|s| !s.is_empty())
+                .map(|s| s.to_string())
+        });
+
+        let filename = filename_from_header
+            .or(filename_from_url)
+            .unwrap_or_else(|| "downloaded_file".to_string());
+
+        info!("File name: {}", filename);
+
         let bytes = response.bytes().await?;
 
-        let ext = if bytes.starts_with(b"PGDMP") {
-            // Postgres custom format
-            "dump"
-        } else if bytes.starts_with(&[0x1F, 0x8B]) {
-            // gzip compressed -> could be Postgres directory dump or MySQL gzipped SQL
-            "tar.gz"
-        } else if bytes.starts_with(b"--") || bytes.starts_with(b"/*") {
-            // Plain MySQL SQL dump
-            "sql"
+        let is_legacy_file = if filename.ends_with(".sql") {
+            true
+        } else if filename.ends_with(".dump") {
+            true
         } else {
-            // Fallback generic
-            "dump"
+            false
         };
+        let downloaded_file = tmp_path.join(&filename);
+        tokio::fs::write(&downloaded_file, &bytes).await?;
+        info!("Backup downloaded to {}", downloaded_file.display());
 
-        info!("Backup dump from {} to {}", tmp_path.display(), ext);
+        let backup_file_path: PathBuf = if !is_legacy_file {
+            let encrypted = if filename.ends_with(".tar.gz") {
+                false
+            } else if filename.ends_with(".tar.gz.enc") {
+                true
+            } else {
+                return Ok(RestoreResult {
+                    generated_id,
+                    status: "failed".into(),
+                });
+            };
 
-        let backup_file_path = tmp_path.join(format!("backup_file_tmp.{}", ext));
-        tokio::fs::write(&backup_file_path, &bytes).await?;
-        info!("Backup downloaded to {}", backup_file_path.display());
+            info!("Encrypted: {}", encrypted);
+
+            let mut compressed_archive = downloaded_file.clone();
+
+            if encrypted {
+                let new_name = downloaded_file
+                    .file_name()
+                    .and_then(|n| n.to_str())
+                    .and_then(|n| n.strip_suffix(".enc"))
+                    .ok_or_else(|| anyhow::anyhow!("Invalid encrypted filename"))?;
+
+                let new_compressed_archive = tmp_path.join(new_name);
+
+                decrypt_file_stream_gcm(
+                    downloaded_file,
+                    new_compressed_archive.clone(),
+                    ctx.edge_key.master_key_b64.clone(),
+                )
+                .await
+                .map_err(|e| {
+                    error!("Failed to decrypt file: {}", e);
+                    e
+                })?;
+
+                compressed_archive = new_compressed_archive;
+            }
+
+            let decompressed_files =
+                decompress_large_tar_gz(compressed_archive.as_path(), tmp_path).await?;
+
+            if decompressed_files.is_empty() {
+                return Ok(RestoreResult {
+                    generated_id,
+                    status: "failed".into(),
+                });
+            }
+
+            if decompressed_files.len() == 1 {
+                decompressed_files[0].clone()
+            } else {
+                compressed_archive
+            }
+        } else {
+            downloaded_file.clone()
+        };
 
         let db_instance = DatabaseFactory::create_for_restore(cfg.clone(), &backup_file_path).await;
         let reachable = db_instance.ping().await.unwrap_or(false);
@@ -115,7 +198,7 @@ impl RestoreService {
                 status: "success".into(),
             }),
             Err(e) => {
-                log::error!("Restore failed: {:?}", e);
+                error!("Restore failed: {:?}", e);
                 Ok(RestoreResult {
                     generated_id,
                     status: "failed".into(),
@@ -124,6 +207,7 @@ impl RestoreService {
         }
     }
 
+    // TODO : update with ctx api manager
     pub async fn send_result(&self, result: RestoreResult) {
         info!(
             "[RestoreService] DB: {} | Status: {}",
@@ -147,7 +231,7 @@ impl RestoreService {
                 if status.is_success() {
                     info!("Restoration result sent successfully");
                 } else {
-                    let text = resp.text().await.unwrap_or_default(); // consumes resp
+                    let text = resp.text().await.unwrap_or_default();
                     error!(
                         "Restoration result failed, status: {}, body: {}",
                         status, text

src/services/status.rs

@@ -4,68 +4,11 @@ use crate::core::context::Context;
 use crate::services::config::DatabaseConfig;
 use crate::settings::CONFIG;
 use reqwest::Client;
-use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::sync::Arc;
-use tracing::{error, info};
+use crate::services::api::endpoints::status::DatabasePayload;
+use crate::services::api::models::agent::status::PingResult;
 
-/// Payload for sending database info in the request
-#[derive(Serialize)]
-struct DatabasePayload<'a> {
-    name: &'a str,
-    dbms: &'a str,
-    #[serde(rename = "generatedId")]
-    generated_id: &'a str,
-}
-
-/// Body for the status API request
-#[derive(Serialize)]
-struct StatusRequestBody<'a> {
-    version: &'a str,
-    databases: Vec<DatabasePayload<'a>>,
-}
-
-/// Typed structs for the response
-#[derive(Debug, Deserialize)]
-pub struct PingResult {
-    pub agent: AgentInfo,
-    pub databases: Vec<DatabaseStatus>,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct AgentInfo {
-    pub id: String,
-    #[serde(rename = "lastContact")]
-    pub last_contact: String,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct DatabaseStatus {
-    pub dbms: String,
-    #[serde(rename = "generatedId")]
-    pub generated_id: String,
-    pub data: DatabaseData,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct DatabaseData {
-    pub backup: BackupInfo,
-    pub restore: RestoreInfo,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct BackupInfo {
-    pub action: bool,
-    pub cron: Option<String>, // can be null
-}
-
-#[derive(Debug, Deserialize)]
-pub struct RestoreInfo {
-    pub action: bool,
-    pub file: String,
-}
-
-/// Service for contacting the agent API
 pub struct StatusService {
     ctx: Arc<Context>,
     client: Client,
@@ -92,27 +35,7 @@ impl StatusService {
             .collect();
 
         let version_str = CONFIG.app_version.as_str();
-
-        let body = StatusRequestBody {
-            version: &version_str,
-            databases: databases_payload,
-        };
-
-        let url = format!(
-            "{}/api/agent/{}/status",
-            edge_key.server_url, edge_key.agent_id
-        );
-        info!("Status request | {}", url);
-
-        let resp = self.client.post(&url).json(&body).send().await?;
-        if !resp.status().is_success() {
-            let msg = format!("Request failed with status: {}", resp.status());
-            error!("{}", msg);
-            return Err(msg.into());
-        }
-
-        let result: PingResult = resp.json().await?;
-
+        let result = self.ctx.api.agent_status(&edge_key.agent_id, &version_str, databases_payload).await?.unwrap();
         Ok(result)
     }
 }

src/services/storage/mod.rs

@@ -0,0 +1,40 @@
+pub mod providers;
+
+use crate::core::context::Context;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::utils::common::BackupMethod;
+use async_trait::async_trait;
+use providers::local;
+use providers::s3;
+use providers::google_drive;
+use std::sync::Arc;
+use tracing::{error, info};
+use crate::services::api::models::agent::status::DatabaseStorage;
+
+#[async_trait]
+pub trait StorageProvider: Send + Sync {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        method: BackupMethod,
+        config: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult;
+}
+
+/// Factory to create provider instance from storage config
+pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider>> {
+    info!("Getting provider");
+    info!("{:#?}", storage.provider.as_str());
+
+    match storage.provider.as_str() {
+        "local" => Some(Box::new(local::LocalProvider {})),
+        "s3" => Some(Box::new(s3::S3Provider {})),
+        "google-drive" => Some(Box::new(google_drive::GoogleDriveProvider {})),
+        _ => {
+            error!("Unknown storage provider: {}", storage.provider);
+            None
+        }
+    }
+}

src/services/storage/providers/google_drive/helpers.rs

@@ -0,0 +1,272 @@
+use anyhow::{Context, Result, anyhow};
+use futures::StreamExt;
+use oauth2::{
+    AuthUrl, ClientId, ClientSecret, RefreshToken, TokenResponse, TokenUrl, basic::BasicClient,
+    reqwest::Client as OAuth2ReqwestClient,
+};
+use reqwest::{Client as ReqwestClient, StatusCode};
+use serde_json::{Value, json};
+use futures::{Stream};
+use bytes::Bytes;
+use reqwest::{Client, header};
+use crate::services::storage::providers::google_drive::models::GoogleDriveProviderConfig;
+
+pub async fn get_google_drive_token(config: &GoogleDriveProviderConfig) -> Result<String> {
+    let http_client = OAuth2ReqwestClient::new();
+
+    let oauth_client = BasicClient::new(ClientId::new(config.client_id.clone()))
+        .set_client_secret(ClientSecret::new(config.client_secret.clone()))
+        .set_auth_uri(
+            AuthUrl::new("https://accounts.google.com/o/oauth2/auth".to_string())
+                .context("invalid auth uri")?,
+        )
+        .set_token_uri(
+            TokenUrl::new("https://oauth2.googleapis.com/token".to_string())
+                .context("invalid token uri")?,
+        );
+
+    let token_result = oauth_client
+        .exchange_refresh_token(&RefreshToken::new(config.refresh_token.clone()))
+        .request_async(&http_client)
+        .await
+        .context("failed to exchange refresh token")?;
+
+    Ok(token_result.access_token().secret().clone())
+}
+
+pub async fn ensure_folder_path(config: &GoogleDriveProviderConfig, path_parts: &[&str]) -> Result<String> {
+    if path_parts.is_empty() {
+        return Ok(config.folder_id.clone());
+    }
+
+    let token = get_google_drive_token(config).await?;
+    let client = ReqwestClient::new();
+    let mut parent_id = config.folder_id.clone();
+
+    for &name in path_parts {
+        let query = format!(
+            "'{parent_id}' in parents and name='{name}' and mimeType='application/vnd.google-apps.folder' and trashed=false"
+        );
+
+        let res = client
+            .get("https://www.googleapis.com/drive/v3/files")
+            .bearer_auth(&token)
+            .query(&[
+                ("q", query),
+                ("fields", "files(id,name)".to_string()),
+                ("supportsAllDrives", "true".to_string()),
+                ("includeItemsFromAllDrives", "true".to_string()),
+                ("corpora", "allDrives".to_string()),
+            ])
+            .send()
+            .await
+            .context("list folders failed")?
+            .json::<Value>()
+            .await?;
+
+        if let Some(files) = res["files"].as_array() {
+            if let Some(folder) = files.first() {
+                if let Some(id) = folder["id"].as_str() {
+                    parent_id = id.to_string();
+                    continue;
+                }
+            }
+        }
+
+        let create_payload = json!({
+            "name": name,
+            "mimeType": "application/vnd.google-apps.folder",
+            "parents": [parent_id],
+            "supportsAllDrives": true,
+        });
+
+        let folder = client
+            .post("https://www.googleapis.com/drive/v3/files")
+            .bearer_auth(&token)
+            .json(&create_payload)
+            .send()
+            .await
+            .context("create folder failed")?
+            .json::<Value>()
+            .await?;
+
+        parent_id = folder["id"]
+            .as_str()
+            .ok_or_else(|| anyhow!("No id returned after folder creation"))?
+            .to_string();
+    }
+
+    Ok(parent_id)
+}
+
+pub async fn find_file_by_name(
+    config: &GoogleDriveProviderConfig,
+    file_name: &str,
+    folder_id: &str,
+) -> Result<Option<String>> {
+    let token = get_google_drive_token(config).await?;
+    let client = ReqwestClient::new();
+
+    let query = format!("'{folder_id}' in parents and name='{file_name}' and trashed=false");
+
+    let res = client
+        .get("https://www.googleapis.com/drive/v3/files")
+        .bearer_auth(&token)
+        .query(&[
+            ("q", query),
+            ("fields", "files(id,name)".to_string()),
+            ("supportsAllDrives", "true".to_string()),
+            ("includeItemsFromAllDrives", "true".to_string()),
+            ("corpora", "allDrives".to_string()),
+        ])
+        .send()
+        .await?
+        .json::<Value>()
+        .await?;
+
+    if let Some(files) = res["files"].as_array() {
+        if let Some(file) = files.first() {
+            if let Some(id) = file["id"].as_str() {
+                return Ok(Some(id.to_string()));
+            }
+        }
+    }
+
+    Ok(None)
+}
+
+pub async fn upload_stream_to_google_drive(
+    config: &GoogleDriveProviderConfig,
+    full_path: &str,
+    mut content_stream: impl Stream<Item = Result<Bytes, std::io::Error>> + Send + Unpin + 'static,
+    total_size: u64,
+    mime_type: Option<&str>,
+) -> Result<()> {
+    let path_parts: Vec<&str> = full_path.split('/').filter(|s| !s.is_empty()).collect();
+    if path_parts.is_empty() {
+        return Err(anyhow::anyhow!("Invalid path: empty"));
+    }
+
+    let file_name = *path_parts.last().unwrap();
+    let folder_path = &path_parts[..path_parts.len() - 1];
+
+    let folder_id = ensure_folder_path(config, folder_path).await?;
+
+    if find_file_by_name(config, file_name, &folder_id).await?.is_some() {
+        return Err(anyhow::anyhow!("File already exists: {}", full_path));
+    }
+
+    let token = get_google_drive_token(config).await?;
+    let client = Client::new();
+
+    let mime = mime_type.unwrap_or("application/octet-stream");
+
+    let metadata = json!({
+        "name": file_name,
+        "parents": [folder_id],
+        "mimeType": mime,
+        "supportsAllDrives": true,
+    });
+
+    let session_res = client
+        .post("https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable")
+        .bearer_auth(&token)
+        .header("X-Upload-Content-Type", mime)
+        .header("X-Upload-Content-Length", total_size.to_string())  // Helps a lot
+        .json(&metadata)
+        .send()
+        .await
+        .context("Failed to initiate resumable upload")?;
+
+    if session_res.status() != StatusCode::OK {
+        let text = session_res.text().await.unwrap_or_default();
+        return Err(anyhow::anyhow!("Initiate failed: {}", text));
+    }
+
+    let upload_url = session_res
+        .headers()
+        .get(header::LOCATION)
+        .ok_or_else(|| anyhow::anyhow!("No Location header"))?
+        .to_str()?
+        .to_string();
+
+    const CHUNK_SIZE: u64 = 8 * 1024 * 1024;
+
+    let mut uploaded: u64 = 0;
+
+    while uploaded < total_size {
+        let chunk_size = (total_size - uploaded).min(CHUNK_SIZE);
+
+        let mut chunk_bytes = Vec::with_capacity(chunk_size as usize);
+        let mut remaining = chunk_size;
+
+        while remaining > 0 {
+            match content_stream.next().await {
+                Some(Ok(bytes)) => {
+                    let to_take = remaining.min(bytes.len() as u64) as usize;
+                    chunk_bytes.extend_from_slice(&bytes[..to_take]);
+                    remaining -= to_take as u64;
+
+                    if to_take < bytes.len() {
+                        // TODO : Put remainder back
+                    }
+                }
+                Some(Err(e)) => return Err(e).context("Stream error during chunk"),
+                None => {
+                    if uploaded + chunk_bytes.len() as u64 != total_size {
+                        return Err(anyhow::anyhow!("Stream ended early"));
+                    }
+                }
+            }
+        }
+
+        if chunk_bytes.is_empty() && uploaded < total_size {
+            return Err(anyhow::anyhow!("Unexpected end of stream"));
+        }
+
+        let range_end = uploaded + chunk_bytes.len() as u64 - 1;
+        let content_range = if uploaded + chunk_bytes.len() as u64 == total_size {
+            format!("bytes {}-{}/{}", uploaded, range_end, total_size)
+        } else {
+            format!("bytes {}-{}/*", uploaded, range_end)
+        };
+
+        let mut retries = 0;
+        loop {
+            let res = client
+                .put(&upload_url)
+                .header("Content-Range", &content_range)
+                .header("Content-Length", chunk_bytes.len().to_string())
+                .body(chunk_bytes.clone())  // clone is cheap if small; optimize later if needed
+                .send()
+                .await;
+
+            match res {
+                Ok(resp) if resp.status().is_success() || resp.status() == StatusCode::PERMANENT_REDIRECT => {
+                    // 200 or 308 = good
+                    uploaded += chunk_bytes.len() as u64;
+                    tracing::info!("Uploaded {}/{} bytes", uploaded, total_size);
+                    break;
+                }
+                Ok(resp) if resp.status() == StatusCode::TOO_MANY_REQUESTS => {
+                    // Backoff on 429
+                    tokio::time::sleep(std::time::Duration::from_secs(5 * (1 << retries))).await;
+                }
+                Ok(resp) => {
+                    let text = resp.text().await.unwrap_or_default();
+                    return Err(anyhow::anyhow!("Chunk upload failed: {}", text));
+                }
+                Err(e) if e.is_timeout() || e.is_connect() => {
+                    if retries > 5 {
+                        return Err(e).context("Too many retries");
+                    }
+                    retries += 1;
+                    tokio::time::sleep(std::time::Duration::from_secs(2u64.pow(retries))).await;
+                }
+                Err(e) => return Err(e).context("Chunk request failed"),
+            }
+        }
+    }
+
+    Ok(())
+}

src/services/storage/providers/google_drive/mod.rs

@@ -0,0 +1,129 @@
+mod helpers;
+mod models; 
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use async_trait::async_trait;
+use std::sync::Arc;
+use tokio::fs;
+use tracing::{error, info};
+use crate::services::storage::providers::google_drive::helpers::{upload_stream_to_google_drive};
+use crate::services::storage::providers::google_drive::models::GoogleDriveProviderConfig;
+
+pub struct GoogleDriveProvider {}
+
+#[async_trait]
+impl StorageProvider for GoogleDriveProvider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        _method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("Missing backup file path".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64
+        )
+            .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let config: GoogleDriveProviderConfig = match storage.clone().config.try_into() {
+            Ok(c) => c,
+            Err(e) => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+
+        let file_name = full_file_name(encrypt);
+
+        info!("Uploading file {}", file_name);
+
+        let remote_file_path = full_file_path(&file_name);
+
+        match upload_stream_to_google_drive(
+            &config,
+            &remote_file_path,
+            upload.stream,   
+            total_size,
+            Some("application/octet-stream"),
+        ).await {
+            Ok(_file_id) => {
+                
+                info!("Google Drive upload successful");
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: true,
+                    error: None,
+                    remote_file_path: Some(remote_file_path),
+                    total_size: Some(total_size),
+                }
+            }
+            Err(e) => {
+                error!("Google Drive upload failed: {:?}", e);
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: Some(total_size),
+                }
+            }
+        }
+
+    }
+}

src/services/storage/providers/google_drive/models.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct GoogleDriveProviderConfig {
+    pub client_id: String,
+    pub client_secret: String,
+    pub refresh_token: String,
+    pub folder_id: String,
+}
+
+

src/services/storage/providers/local.rs

@@ -0,0 +1,116 @@
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use crate::utils::tus::upload_to_tus_stream_with_headers;
+use async_trait::async_trait;
+use reqwest::header::{HeaderMap, HeaderValue};
+use std::sync::Arc;
+use tokio::fs;
+use tracing::error;
+
+pub struct LocalProvider;
+
+#[async_trait]
+impl StorageProvider for LocalProvider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("File path error".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let file_name = full_file_name(encrypt);
+        let remote_file_path = full_file_path(&file_name);
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64
+        )
+        .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None
+                };
+            }
+        };
+
+        let mut extra_headers = HeaderMap::new();
+
+        extra_headers.insert("X-File-Name", HeaderValue::from_str(&file_name).unwrap());
+        extra_headers.insert("X-File-Size", HeaderValue::from_str(&total_size.to_string()).unwrap());
+        extra_headers.insert(
+            "X-File-Path",
+            HeaderValue::from_str(&remote_file_path).unwrap(),
+        );
+        extra_headers.insert(
+            "X-Generated-Id",
+            HeaderValue::from_str(&result.generated_id).unwrap(),
+        );
+        extra_headers.insert("X-Status", HeaderValue::from_str(&result.status).unwrap());
+        extra_headers.insert(
+            "X-Method",
+            HeaderValue::from_str(&method.to_string()).unwrap(),
+        );
+        
+        let tus_endpoint = format!("{}/tus/files", ctx.edge_key.server_url);
+
+        match upload_to_tus_stream_with_headers(upload.stream, &tus_endpoint, extra_headers, total_size).await {
+            Ok(_) => UploadResult {
+                storage_id: storage.id.clone(),
+                success: true,
+                error: None,
+                remote_file_path: Some(remote_file_path),
+                total_size: Some(total_size),
+            },
+            Err(e) => {
+                error!("Local upload failed: {}", e);
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                }
+            }
+        }
+    }
+}

src/services/storage/providers/mod.rs

@@ -0,0 +1,3 @@
+pub mod local;
+pub mod s3;
+pub mod google_drive;

src/services/storage/providers/s3/mod.rs

@@ -0,0 +1,314 @@
+mod models;
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::services::storage::providers::s3::models::S3ProviderConfig;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use async_trait::async_trait;
+use aws_sdk_s3 as s3;
+use aws_sdk_s3::config::BehaviorVersion;
+use aws_sdk_s3::config::Region;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+use futures::StreamExt;
+use std::pin::Pin;
+use std::sync::Arc;
+use tokio::fs;
+use tracing::{error, info};
+
+pub struct S3Provider {}
+
+#[async_trait]
+impl StorageProvider for S3Provider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        _method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("Missing backup file path".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64,
+        )
+        .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let config: S3ProviderConfig = match storage.clone().config.try_into() {
+            Ok(c) => c,
+            Err(e) => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let credentials = s3::config::Credentials::new(
+            config.access_key.clone(),
+            config.secret_key.clone(),
+            None,
+            None,
+            "static-creds",
+        );
+
+        let region = Region::new(config.region.clone().unwrap_or("us-east-1".to_string()));
+
+        let sdk_config = s3::config::Builder::new()
+            .credentials_provider(credentials)
+            .region(region)
+            .force_path_style(true)
+            .endpoint_url(format!(
+                "{}://{}",
+                if config.ssl { "https" } else { "http" },
+                config.end_point_url
+            ))
+            .behavior_version(BehaviorVersion::latest())
+            .build();
+
+        let client = s3::Client::from_conf(sdk_config);
+
+        const PART_SIZE: usize = 100 * 1024 * 1024; // 100 MiB
+
+        let file_name = full_file_name(encrypt);
+
+        info!("Uploading file {}", file_name);
+
+        let bucket = &config.bucket_name;
+        let remote_file_path = full_file_path(&file_name);
+        info!("S3 key {:}", remote_file_path);
+        info!(
+            "Starting multipart upload to s3://{}/{}",
+            bucket, remote_file_path
+        );
+
+        let create_resp = match client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(&remote_file_path)
+            .send()
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                error!("Failed to create multipart upload: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let upload_id = match create_resp.upload_id {
+            Some(id) => id,
+            None => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some("No upload ID returned".to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let mut parts: Vec<CompletedPart> = Vec::new();
+        let mut part_number: i32 = 1;
+        let mut buffer: Vec<u8> = Vec::with_capacity(PART_SIZE);
+
+        let mut peekable = upload.stream.peekable();
+
+        while let Some(item) = peekable.next().await {
+            let bytes = match item {
+                Ok(b) => b,
+                Err(e) => {
+                    error!("Stream error during upload: {}", e);
+                    let _ = client
+                        .abort_multipart_upload()
+                        .bucket(bucket)
+                        .key(&remote_file_path)
+                        .upload_id(&upload_id)
+                        .send()
+                        .await;
+                    return UploadResult {
+                        storage_id: storage.id.clone(),
+                        success: false,
+                        error: Some(format!("Stream error: {}", e)),
+                        remote_file_path: None,
+                        total_size: None,
+                    };
+                }
+            };
+
+            buffer.extend_from_slice(&bytes);
+
+            let is_last = {
+                let pinned = Pin::new(&mut peekable);
+                let peek_future = pinned.peek();
+                peek_future.await.is_none()
+            };
+
+            let should_upload = buffer.len() >= PART_SIZE || is_last;
+
+            if should_upload && !buffer.is_empty() {
+                let body = ByteStream::from(buffer.clone());
+
+                match client
+                    .upload_part()
+                    .bucket(bucket)
+                    .key(&remote_file_path)
+                    .upload_id(&upload_id)
+                    .part_number(part_number)
+                    .body(body)
+                    .send()
+                    .await
+                {
+                    Ok(resp) => {
+                        if let Some(etag) = resp.e_tag {
+                            parts.push(
+                                CompletedPart::builder()
+                                    .part_number(part_number)
+                                    .e_tag(etag)
+                                    .build(),
+                            );
+                            info!("Uploaded part {} ({} bytes)", part_number, buffer.len());
+                        }
+                    }
+                    Err(e) => {
+                        error!("Failed to upload part {}: {}", part_number, e);
+                        let _ = client
+                            .abort_multipart_upload()
+                            .bucket(bucket)
+                            .key(&remote_file_path)
+                            .upload_id(&upload_id)
+                            .send()
+                            .await;
+                        return UploadResult {
+                            storage_id: storage.id.clone(),
+                            success: false,
+                            error: Some(e.to_string()),
+                            remote_file_path: None,
+                            total_size: None,
+                        };
+                    }
+                }
+                buffer.clear();
+                part_number += 1;
+            }
+        }
+
+        if !buffer.is_empty() {
+            let _ = client
+                .abort_multipart_upload()
+                .bucket(bucket)
+                .key(&remote_file_path)
+                .upload_id(&upload_id)
+                .send()
+                .await;
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("No parts were uploaded".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        }
+
+        let completed = CompletedMultipartUpload::builder()
+            .set_parts(Some(parts))
+            .build();
+
+        match client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(&remote_file_path)
+            .upload_id(&upload_id)
+            .multipart_upload(completed)
+            .send()
+            .await
+        {
+            Ok(_) => {
+                info!(
+                    "Successfully completed multipart upload: {}",
+                    remote_file_path
+                );
+
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: true,
+                    error: None,
+                    remote_file_path: Some(remote_file_path),
+                    total_size: Some(total_size),
+                }
+            }
+            Err(e) => {
+                error!("Failed to complete multipart upload: {}", e);
+                let _ = client
+                    .abort_multipart_upload()
+                    .bucket(bucket)
+                    .key(&remote_file_path)
+                    .upload_id(&upload_id)
+                    .send()
+                    .await;
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                }
+            }
+        }
+    }
+}

src/services/storage/providers/s3/models.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct S3ProviderConfig {
+    pub access_key: String,
+    pub secret_key: String,
+    pub bucket_name: String,
+    pub end_point_url: String,
+    pub ssl: bool,
+    pub region: Option<String>,
+}

src/tasks/ping.rs

@@ -19,33 +19,6 @@ pub async fn ping_server() {
                 e
             );
         }
-
         tokio::time::sleep(Duration::from_secs(CONFIG.pooling as u64)).await;
     }
-}
-// use crate::core::agent::Agent;
-// use crate::core::context::Context;
-// use crate::utils::common::BackupMethod;
-// use std::sync::Arc;
-// use tokio::sync::Mutex;
-// use tokio::time::{sleep, Duration};
-// use tracing::{error};
-//
-// pub async fn ping_server() {
-//
-//
-//     loop {
-//         let ctx = Arc::new(Context::new());
-//         let agent = Arc::new(Mutex::new(Agent::new(ctx.clone()).await));
-//         let agent_clone = agent.clone();
-//
-//         tokio::spawn(async move {
-//             let mut agent_locked = agent_clone.lock().await;
-//             if let Err(e) = agent_locked.run(BackupMethod::Manual).await {
-//                 error!("An error occurred while executing ping_server: {:?}", e);
-//             }
-//         });
-//
-//         sleep(Duration::from_secs(5)).await;
-//     }
-// }
+}

src/utils/common.rs

@@ -1,3 +1,6 @@
+use serde::Serialize;
+use serde_json::Value;
+
 #[derive(Clone, Copy)]
 pub enum BackupMethod {
     Automatic,
@@ -12,3 +15,12 @@ impl ToString for BackupMethod {
         }
     }
 }
+
+
+pub fn vec_to_option_json<T: Serialize>(v: Vec<T>) -> Option<Value> {
+    if v.is_empty() {
+        None
+    } else {
+        Some(serde_json::to_value(v).expect("serialization failed"))
+    }
+}

src/utils/compress.rs

@@ -0,0 +1,97 @@
+use anyhow::Result;
+use async_compression::tokio::bufread::GzipDecoder;
+use async_compression::tokio::write::GzipEncoder as AsyncGzipEncoder;
+use futures::StreamExt;
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::fs::{create_dir_all};
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio_tar::Archive;
+use tokio_tar::Builder as TokioTarBuilder;
+use tracing::info;
+
+#[allow(dead_code)]
+pub struct CompressionResult {
+    pub compressed_path: PathBuf,
+}
+
+pub async fn compress_to_tar_gz_large(file: &PathBuf) -> Result<CompressionResult> {
+    if file
+        .file_name()
+        .and_then(|n| n.to_str())
+        .map(|n| n.ends_with(".tar.gz"))
+        .unwrap_or(false)
+    {
+        info!("File {:?} is already a tar.gz, skipping compression", file);
+        return Ok(CompressionResult {
+            compressed_path: file.clone(),
+        });
+    }
+
+    let tar_gz_path = file.with_extension("").with_extension("tar.gz");
+
+    let output_file = File::create(&tar_gz_path).await?;
+    let gzip_writer = AsyncGzipEncoder::new(output_file);
+    let mut tar_builder = TokioTarBuilder::new(gzip_writer);
+
+    let file_name = file
+        .file_name()
+        .ok_or_else(|| anyhow::anyhow!("Cannot get file name for {:?}", file))?;
+
+    tar_builder
+        .append_path_with_name(file, file_name)
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to append path: {}", e))?;
+
+    tar_builder
+        .finish()
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to finish tar: {}", e))?;
+
+    let mut gzip = tar_builder
+        .into_inner()
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to extract gzip encoder: {}", e))?;
+
+    gzip.shutdown()
+        .await
+        .map_err(|e| anyhow::anyhow!("Gzip shutdown failed: {}", e))?;
+
+    info!("Compressing {:?} to {:?}", &file, &tar_gz_path);
+
+    Ok(CompressionResult {
+        compressed_path: tar_gz_path,
+    })
+}
+
+pub async fn decompress_large_tar_gz(
+    tar_gz_path: &Path,
+    output_dir: &Path,
+) -> Result<Vec<PathBuf>> {
+    let file = File::open(tar_gz_path).await?;
+    let buf_reader = BufReader::with_capacity(8 * 1024 * 1024, file);
+    let decoder = GzipDecoder::new(buf_reader);
+    let mut archive = Archive::new(decoder);
+
+    let mut extracted_files = Vec::new();
+    let mut entries = archive.entries()?;
+
+    while let Some(entry) = entries.next().await {
+        let mut entry = entry?;
+        let path = entry.path()?.to_path_buf();
+        let full_path = output_dir.join(&path);
+
+        if let Some(parent) = full_path.parent() {
+            create_dir_all(parent).await?;
+        }
+
+        entry.unpack(&full_path).await?;
+        extracted_files.push(full_path);
+    }
+
+    // remove_file(tar_gz_path).await?;
+    info!("Decompressed {:?} into {:?}", tar_gz_path, output_dir);
+
+    Ok(extracted_files)
+}

src/utils/deserializer.rs

@@ -0,0 +1,39 @@
+use serde::{Deserialize, Deserializer};
+use toml::Value;
+
+pub fn deserialize_snake_case<'de, D>(deserializer: D) -> Result<Value, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    Ok(to_snake_case(value))
+}
+fn to_snake_case(value: Value) -> Value {
+    match value {
+        Value::Table(table) => Value::Table(
+            table
+                .into_iter()
+                .map(|(k, v)| (camel_to_snake(&k), to_snake_case(v)))
+                .collect(),
+        ),
+        Value::Array(arr) => {
+            Value::Array(arr.into_iter().map(to_snake_case).collect())
+        }
+        other => other,
+    }
+}
+
+fn camel_to_snake(s: &str) -> String {
+    let mut out = String::new();
+    for (i, c) in s.chars().enumerate() {
+        if c.is_uppercase() {
+            if i > 0 {
+                out.push('_');
+            }
+            out.push(c.to_ascii_lowercase());
+        } else {
+            out.push(c);
+        }
+    }
+    out
+}

src/utils/edge_key.rs

@@ -2,7 +2,6 @@ use base64::{Engine as _, engine::general_purpose};
 use serde::Deserialize;
 use serde_json::Value;
 use thiserror::Error;
-use tracing::error;
 
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
@@ -11,8 +10,8 @@ pub struct EdgeKey {
     pub server_url: String,
     #[serde(rename = "agentId")]
     pub agent_id: String,
-    #[serde(rename = "publicKey")]
-    pub public_key: String,
+    #[serde(rename = "masterKeyB64")]
+    pub master_key_b64: String,
 }
 
 #[derive(Debug, Error)]
@@ -35,14 +34,14 @@ pub fn decode_edge_key(edge_key: &str) -> Result<EdgeKey, EdgeKeyError> {
     let decoded_str = String::from_utf8_lossy(&decoded_bytes);
 
     let parsed: Value = serde_json::from_str(&decoded_str)?;
+    
     if parsed.get("serverUrl").is_some()
         && parsed.get("agentId").is_some()
-        && parsed.get("publicKey").is_some()
+        && parsed.get("masterKeyB64").is_some()
     {
         let edge_key: EdgeKey = serde_json::from_value(parsed)?;
         Ok(edge_key)
     } else {
-        error!("EDGE_KEY INVALID");
         Err(EdgeKeyError::InvalidKey)
     }
 }

src/utils/file.rs

@@ -1,15 +1,175 @@
-use std::path::Path;
+#![allow(dead_code)]
+
+use anyhow::Result;
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
+use uuid::Uuid;
+
+use aes_gcm::aead::{Aead, KeyInit};
+use aes_gcm::{Aes256Gcm, Key, Nonce};
+use base64::Engine;
+use base64::engine::general_purpose;
+use bytes::Bytes;
+use futures::Stream;
+use rand::rngs::OsRng;
+use rand::TryRngCore;
+use tokio::io::{AsyncWriteExt, BufWriter};
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::info;
+
+#[derive(Serialize, Deserialize)]
+pub struct EncryptionMetadataFile {
+    pub version: u8,
+    pub cipher: String,
+    pub encrypted_aes_key_b64: String,
+    pub iv_b64: String,
+}
 
 pub fn full_extension(path: &Path) -> String {
     path.file_name()
-        .and_then(|name| name.to_str())
-        .map(|name| {
-            match name.find('.') {
-                Some(idx) => &name[idx..],
-                None => "",
-            }
-        })
-        .unwrap_or_default()
+        .and_then(|n| n.to_str())
+        .and_then(|n| n.find('.').map(|i| &n[i..]))
+        .unwrap_or("")
         .to_string()
 }
 
+pub fn full_file_name(encrypt: bool) -> String {
+    let uuid = Uuid::new_v4();
+    let base_name = format!("{}.{}", uuid, "tar.gz");
+    if encrypt {
+        format!("{}.enc", base_name)
+    } else {
+        base_name.to_string()
+    }
+}
+
+pub fn full_file_path(file_name: &String) -> String {
+    format!("backups/{}/{}", Utc::now().format("%Y-%m-%d"), file_name)
+}
+
+const CHUNK_SIZE: usize = 16 * 1024 * 1024;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct FileHeader {
+    version: u8,
+    cipher: String,
+    chunk_size: usize,
+    base_nonce: Vec<u8>,
+}
+
+pub async fn encrypt_file_stream_gcm(
+    file_path: PathBuf,
+    master_key_b64: String,
+) -> Result<impl Stream<Item = Result<Bytes>> + Send + 'static> {
+    let master_key_bytes = general_purpose::STANDARD
+        .decode(master_key_b64)
+        .map_err(|_| anyhow::anyhow!("Invalid base64"))?;
+
+    let (tx, rx) = mpsc::channel(8);
+
+    tokio::spawn(async move {
+        let mut rng = OsRng;
+        let mut base_nonce = [0u8; 8];
+        rng.try_fill_bytes(&mut base_nonce).unwrap();
+
+        let key = Key::<Aes256Gcm>::try_from(master_key_bytes.as_slice())
+            .map_err(|_| anyhow::anyhow!("Invalid AES-256 key length")).unwrap();
+
+        let cipher = Aes256Gcm::new(&key);
+
+        let header = FileHeader {
+            version: 1,
+            cipher: "AES-256-GCM".to_string(),
+            chunk_size: CHUNK_SIZE,
+            base_nonce: base_nonce.to_vec(),
+        };
+        let header_json = serde_json::to_string(&header).unwrap();
+        tx.send(Ok(Bytes::from(header_json + "\n"))).await.unwrap();
+
+        let file = File::open(&file_path).await.unwrap();
+        let mut reader = BufReader::new(file);
+        let mut buffer = vec![0u8; CHUNK_SIZE];
+        let mut chunk_index: u32 = 0;
+
+        loop {
+            let n = reader.read(&mut buffer).await.unwrap();
+            if n == 0 {
+                break;
+            }
+
+            let mut nonce_bytes = [0u8; 12];
+            nonce_bytes[..8].copy_from_slice(&base_nonce);
+            nonce_bytes[8..].copy_from_slice(&chunk_index.to_be_bytes());
+            let nonce = Nonce::try_from(&nonce_bytes[..])
+                .map_err(|_| anyhow::anyhow!("Invalid nonce length")).unwrap();
+
+            let ciphertext = cipher.encrypt(&nonce, &buffer[..n]).unwrap();
+            let mut out = Vec::with_capacity(4 + ciphertext.len());
+            out.extend_from_slice(&(ciphertext.len() as u32).to_be_bytes());
+            out.extend_from_slice(&ciphertext);
+
+            tx.send(Ok(Bytes::from(out))).await.unwrap();
+            chunk_index += 1;
+        }
+    });
+
+    Ok(ReceiverStream::new(rx))
+}
+
+pub async fn decrypt_file_stream_gcm(
+    encrypted_path: PathBuf,
+    decrypted_path: PathBuf,
+    master_key_b64: String,
+) -> Result<()> {
+    info!("Decrypting {:?}", decrypted_path);
+
+    let master_key_bytes = general_purpose::STANDARD
+        .decode(master_key_b64)
+        .map_err(|_| anyhow::anyhow!("Invalid base64"))?;
+
+    let mut reader = BufReader::new(File::open(&encrypted_path).await?);
+
+    let mut header_line = Vec::new();
+    reader.read_until(b'\n', &mut header_line).await?;
+    let header: FileHeader = serde_json::from_slice(&header_line)?;
+
+    let key = Key::<Aes256Gcm>::try_from(master_key_bytes.as_slice())
+        .map_err(|_| anyhow::anyhow!("Invalid AES-256 key length"))?;
+    let cipher = Aes256Gcm::new(&key);
+
+    let mut writer = BufWriter::new(File::create(&decrypted_path).await?);
+    let mut chunk_index: u32 = 0;
+
+    loop {
+        let mut len_buf = [0u8; 4];
+        match reader.read_exact(&mut len_buf).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
+            Err(e) => return Err(e.into()),
+        }
+        let chunk_len = u32::from_be_bytes(len_buf) as usize;
+
+        let mut chunk_ciphertext = vec![0u8; chunk_len];
+        reader.read_exact(&mut chunk_ciphertext).await?;
+
+        let mut nonce_bytes = [0u8; 12];
+        nonce_bytes[..8].copy_from_slice(&header.base_nonce);
+        nonce_bytes[8..].copy_from_slice(&chunk_index.to_be_bytes());
+        let nonce = Nonce::try_from(&nonce_bytes[..])
+            .map_err(|_| anyhow::anyhow!("Invalid nonce length"))?;
+
+        let plaintext = cipher
+            .decrypt(&nonce, chunk_ciphertext.as_slice())
+            .map_err(|e| anyhow::anyhow!("AES-GCM decryption failed: {:?}", e))?;
+
+        writer.write_all(&plaintext).await?;
+        chunk_index += 1;
+    }
+
+    writer.flush().await?;
+    Ok(())
+}

src/utils/locks.rs

@@ -1,10 +1,10 @@
 use anyhow::{Context, Result};
-use chrono::{Local};
-use tracing::{info, warn, error};
+use chrono::Local;
 use std::path::{Path, PathBuf};
 use std::time::{Duration, SystemTime};
-use tokio::fs::{OpenOptions, metadata, remove_file, create_dir_all, read_dir};
+use tokio::fs::{OpenOptions, create_dir_all, metadata, read_dir, remove_file};
 use tokio::io::AsyncWriteExt;
+use tracing::{error, info, warn};
 
 /// Lock type for logging purposes
 #[derive(Debug, Copy, Clone)]
@@ -59,6 +59,24 @@ impl FileLock {
         Ok(())
     }
 
+    pub async fn is_locked(id: &str) -> Result<bool> {
+        let path = Self::lock_file_path(id);
+
+        if !path.exists() {
+            return Ok(false);
+        }
+
+        let meta = metadata(&path).await?;
+
+        if let Ok(modified) = meta.modified() {
+            let age = SystemTime::now().duration_since(modified)?;
+
+            if age > Duration::from_secs(24 * 60 * 60) {
+                return Ok(false);
+            }
+        }
+        Ok(true)
+    }
 
     /// Acquire a file-based lock
     pub async fn acquire(id: &str, service_name: &str) -> Result<()> {
@@ -88,10 +106,13 @@ impl FileLock {
             .await
             .with_context(|| format!("Failed to create lock file for {}", id))?;
 
-        f.write_all(format!("Service: {}\n", service_name).as_bytes()).await?;
-        f.write_all(format!("PID: {}\n", std::process::id()).as_bytes()).await?;
+        f.write_all(format!("Service: {}\n", service_name).as_bytes())
+            .await?;
+        f.write_all(format!("PID: {}\n", std::process::id()).as_bytes())
+            .await?;
         // f.write_all(format!("Timestamp: {}\n", Utc::now()).as_bytes()).await?;
-        f.write_all(format!("Timestamp: {}\n", Local::now()).as_bytes()).await?;
+        f.write_all(format!("Timestamp: {}\n", Local::now()).as_bytes())
+            .await?;
 
         info!("Successfully acquired lock for {}", id);
         Ok(())
@@ -106,7 +127,10 @@ impl FileLock {
             remove_file(&path).await?;
             info!("Released file lock for {}", id);
         } else {
-            warn!("Attempted to release lock for {}, but file does not exist", id);
+            warn!(
+                "Attempted to release lock for {}, but file does not exist",
+                id
+            );
         }
         Ok(())
     }

src/utils/logging.rs

@@ -28,14 +28,14 @@ pub fn init_logger() {
         .with_ansi(false)
         .with_target(false);
 
-    // let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
-    let env_filter = EnvFilter::new(CONFIG.log.clone());
-
+    let env_filter = EnvFilter::new(format!(
+        "{},aws_sdk_s3=warn,aws_smithy_http=warn,aws_smithy_runtime=warn,aws_smithy_client=warn,aws_smithy_types=warn,reqwest=warn,hyper=warn,tokio=warn,h2=warn",
+        CONFIG.log
+    ));
 
     let term_layer = fmt::layer()
         .with_writer(std::io::stdout)
         .with_timer(timer.clone())
-        // .with_timer(tracing_subscriber::fmt::time::LocalTime::rfc_3339())
         .with_ansi(true)
         .with_target(false)
         .with_filter(env_filter);

src/utils/mod.rs

@@ -6,3 +6,7 @@ pub mod text;
 pub mod file;
 pub mod locks;
 pub mod logging;
+pub mod tus;
+pub mod deserializer;
+pub mod compress;
+pub mod stream;

src/utils/stream.rs

@@ -0,0 +1,37 @@
+use crate::utils::file::encrypt_file_stream_gcm;
+use anyhow::Result;
+use bytes::Bytes;
+use futures::{Stream, StreamExt};
+use std::pin::Pin;
+use tokio_util::io::ReaderStream;
+
+pub struct UploadStream {
+    pub stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
+}
+
+pub async fn build_stream(
+    file_path: &std::path::Path,
+    encrypt: bool,
+    master_key_b64: &String,
+) -> Result<UploadStream> {
+    if encrypt {
+        let encrypted_stream =
+            encrypt_file_stream_gcm(file_path.to_path_buf(), master_key_b64.to_string()).await?;
+
+        let stream = Box::pin(
+            encrypted_stream
+                .map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+        );
+
+        Ok(UploadStream { stream })
+    } else {
+        let file = tokio::fs::File::open(file_path).await?;
+        let reader = ReaderStream::new(file);
+
+        let stream = Box::pin(
+            reader.map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+        );
+
+        Ok(UploadStream { stream })
+    }
+}

src/utils/task_manager/cron.rs

@@ -1,17 +1,17 @@
 use crate::utils::task_manager::models;
 use crate::utils::task_manager::tasks::{remove_task, upsert_task};
 use crate::utils::text::normalize_cron;
-use chrono::{Local};
+use chrono::Local;
 use cron::Schedule;
-use tracing::debug;
 use redis::AsyncCommands;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 use std::str::FromStr;
+use tracing::debug;
 use tracing::info;
 
 pub fn next_run_timestamp(expr: &str) -> i64 {
     let schedule = Schedule::from_str(expr).unwrap();
-    // schedule.upcoming(Utc).next().unwrap().timestamp()
     schedule.upcoming(Local).next().unwrap().timestamp()
 }
 
@@ -21,6 +21,7 @@ pub async fn check_and_update_cron(
     args: Vec<String>,
     task: &str,
     task_name: String,
+    metadata: Option<Value>,
 ) {
     let redis_key = format!("redbeat:{}", task_name);
 
@@ -44,16 +45,31 @@ pub async fn check_and_update_cron(
                 let raw: String = conn.hget(&redis_key, "data").await.unwrap();
                 let stored: models::PeriodicTask = serde_json::from_str(&raw).unwrap();
 
-                if stored.cron != cron {
-                    upsert_task(conn, &task_name, task, &cron, args.clone())
-                        .await
-                        .unwrap_or_else(|e| {
-                            tracing::error!("Failed to update task {}: {:?}", task_name, e);
-                        });
-                    info!("Task {} updated", task_name);
+                let cron_changed = stored.cron != cron;
+                let args_changed = stored.args != args;
+                let metadata_changed = stored.metadata != metadata;
+
+                if cron_changed || args_changed || metadata_changed {
+                    upsert_task(
+                        conn,
+                        &task_name,
+                        task,
+                        &cron,
+                        args.clone(),
+                        metadata,
+                    )
+                    .await
+                    .unwrap_or_else(|e| {
+                        tracing::error!("Failed to update task {}: {:?}", task_name, e);
+                    });
+
+                    info!(
+                        "Task {} updated (cron: {}, args: {}, metadata: {})",
+                        task_name, cron_changed, args_changed, metadata_changed
+                    );
                 }
             } else {
-                upsert_task(conn, &task_name, task, &cron, args)
+                upsert_task(conn, &task_name, task, &cron, args, metadata)
                     .await
                     .unwrap_or_else(|e| {
                         tracing::error!("Failed to create task {}: {:?}", task_name, e);

src/utils/task_manager/models.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PeriodicTask {
@@ -6,4 +7,5 @@ pub struct PeriodicTask {
     pub cron: String,
     pub args: Vec<String>,
     pub enabled: bool,
+    pub metadata: Option<Value>,
 }

src/utils/task_manager/scheduler.rs

@@ -5,23 +5,22 @@ use crate::utils::common::BackupMethod;
 use crate::utils::task_manager::cron::next_run_timestamp;
 use crate::utils::task_manager::models::PeriodicTask;
 use crate::utils::task_manager::tasks::SCHEDULE_KEY;
-use tracing::info;
 use redis::AsyncCommands;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 use std::sync::Arc;
 use tracing::error;
+use tracing::info;
+use crate::services::api::models::agent::status::DatabaseStorage;
 
 pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
     loop {
-        // let now = chrono::Utc::now().timestamp();
         let now = chrono::Local::now().timestamp();
-        // info!("Scheduling task {}", chrono::Local::now());
 
         let due: Vec<String> = conn
             .zrangebyscore(SCHEDULE_KEY, 0, now)
             .await
             .unwrap_or_default();
-
         for key in due {
             let raw: String = conn.hget(&key, "data").await.unwrap();
             let task: PeriodicTask = serde_json::from_str(&raw).unwrap();
@@ -29,35 +28,39 @@ pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
             if !task.enabled {
                 continue;
             }
-
             let task_clone = task.clone();
             let mut conn_clone = conn.clone();
 
             tokio::spawn(async move {
                 info!(
-                    "Executing task={} args={:?}",
-                    task_clone.task, task_clone.args
+                    "Executing task={} args={:?} metadata={:?}",
+                    task_clone.task, task_clone.args, task_clone.metadata
                 );
-
-                // let _ = execute_task(task_clone.task.as_str(), task_clone.args).await;
-
-                if let Err(e) = execute_task(task_clone.task.as_str(), task_clone.args).await {
+                if let Err(e) = execute_task(
+                    task_clone.task.as_str(),
+                    task_clone.args,
+                    task_clone.metadata,
+                )
+                .await
+                {
                     error!(
                         "An error occurred while executing task={} : {:?}",
                         task_clone.task, e
                     );
                 }
-
                 let next_ts = next_run_timestamp(&task_clone.cron);
                 let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap();
             });
         }
-
         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     }
 }
 
-pub async fn execute_task(task: &str, args: Vec<String>) -> Result<(), anyhow::Error> {
+pub async fn execute_task(
+    task: &str,
+    args: Vec<String>,
+    metadata: Option<Value>,
+) -> Result<(), anyhow::Error> {
     match task {
         "tasks.database.periodic_backup" => {
             let generated_id = &args[0];
@@ -69,8 +72,24 @@ pub async fn execute_task(task: &str, args: Vec<String>) -> Result<(), anyhow::E
             let backup_service = BackupService::new(ctx.clone());
             let config = config_service.load(None).unwrap();
 
+            let metadata_obj = metadata
+                .into_iter()
+                .next()
+                .ok_or_else(|| anyhow::anyhow!("Metadata missing"))?;
+
+            let storages_value: &Value = metadata_obj
+                .get("storages")
+                .ok_or_else(|| anyhow::anyhow!("storages key missing"))?;
+            
+            let encrypt_value: &Value = metadata_obj
+                .get("encrypt")
+                .ok_or_else(|| anyhow::anyhow!("encrypt key missing"))?;
+            
+            let storages: Vec<DatabaseStorage> = serde_json::from_value(storages_value.clone())?;
+            let encrypt : bool = serde_json::from_value(encrypt_value.clone())?;
+
             backup_service
-                .dispatch(generated_id, &config, BackupMethod::Automatic)
+                .dispatch(generated_id, &config, BackupMethod::Automatic, &storages, encrypt)
                 .await;
 
             Ok(())

src/utils/task_manager/tasks.rs

@@ -3,6 +3,7 @@
 use crate::utils::task_manager::cron::next_run_timestamp;
 use crate::utils::task_manager::models::PeriodicTask;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 
 pub const SCHEDULE_KEY: &str = "redbeat:schedule";
 
@@ -12,6 +13,7 @@ pub async fn upsert_task(
     task: &str,
     cron: &str,
     args: Vec<String>,
+    metadata: Option<Value>,
 ) -> redis::RedisResult<()> {
     let key = format!("redbeat:{}", name);
     let next_ts = next_run_timestamp(cron);
@@ -21,6 +23,7 @@ pub async fn upsert_task(
         cron: cron.to_string(),
         args,
         enabled: true,
+        metadata
     };
 
     let payload = serde_json::to_string(&entry).unwrap();

src/utils/tus.rs

@@ -0,0 +1,98 @@
+use anyhow::Result;
+use bytes::Bytes;
+use futures::{Stream, StreamExt};
+use log::info;
+use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue};
+
+const PATCH_CHUNK_SIZE: usize = 1 * 1024 * 1024;
+
+pub async fn upload_to_tus_stream_with_headers<S>(
+    encrypted_stream: S,
+    tus_endpoint: &str,
+    extra_headers: HeaderMap,
+    total_size: u64,
+) -> Result<()>
+where
+    S: Stream<Item = Result<Bytes, std::io::Error>> + Send + 'static,
+{
+    let client = reqwest::Client::new();
+
+    info!("File size: {}", total_size);
+
+    let mut headers = HeaderMap::new();
+    headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+    headers.insert("Upload-Defer-Length", HeaderValue::from_static("1"));
+
+    let resp = client
+        .post(tus_endpoint)
+        .headers(headers.clone())
+        .send()
+        .await?;
+    if !resp.status().is_success() {
+        anyhow::bail!("Failed to create upload: {}", resp.status());
+    }
+
+    let upload_url = resp
+        .headers()
+        .get("Location")
+        .ok_or_else(|| anyhow::anyhow!("Missing Location header"))?
+        .to_str()?
+        .to_string();
+
+    let mut stream = Box::pin(
+        encrypted_stream.map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+    );
+
+    let mut offset: u64 = 0;
+
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk?;
+        for sub_chunk in chunk.chunks(PATCH_CHUNK_SIZE) {
+            let mut patch_headers = extra_headers.clone();
+            patch_headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+            patch_headers.insert("Upload-Offset", HeaderValue::from_str(&offset.to_string())?);
+            patch_headers.insert(
+                CONTENT_TYPE,
+                HeaderValue::from_static("application/offset+octet-stream"),
+            );
+
+            let patch_resp = client
+                .patch(&upload_url)
+                .headers(patch_headers)
+                .body(sub_chunk.to_vec())
+                .send()
+                .await?;
+
+            if !patch_resp.status().is_success() {
+                anyhow::bail!(
+                    "Chunk upload failed at offset {}: {}",
+                    offset,
+                    patch_resp.status()
+                );
+            }
+            offset += sub_chunk.len() as u64;
+            // info!("Progress: {}/{}", offset, total_size);
+        }
+    }
+
+    let mut finalize_headers = extra_headers.clone();
+    finalize_headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+    finalize_headers.insert("Upload-Offset", HeaderValue::from_str(&offset.to_string())?);
+    finalize_headers.insert("Upload-Length", HeaderValue::from_str(&offset.to_string())?);
+    finalize_headers.insert(
+        CONTENT_TYPE,
+        HeaderValue::from_static("application/offset+octet-stream"),
+    );
+
+    let finalize_resp = client
+        .patch(&upload_url)
+        .headers(finalize_headers)
+        .send()
+        .await?;
+
+    if !finalize_resp.status().is_success() {
+        anyhow::bail!("Failed to finalize upload");
+    }
+
+    Ok(())
+}