Feat/delete files

.github/workflows/build-and-deploy-consumer.yaml

@@ -0,0 +1,55 @@
+name: Build and Deploy Consumer to Production
+on:
+  push:
+    branches: ["master", "feature/cd", "feat/delete-files"]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check Out Repo
+        uses: actions/checkout@v4
+      - name: Login to DockerHub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+      - name: Build and push to DockerHub
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: ${{ secrets.DOCKERHUB_USERNAME }}/background-tasks:${{ github.sha }}
+  deploy:
+    needs: build
+    runs-on: ubuntu-latest
+    environment:
+      name: production
+    steps:
+      - uses: actions/checkout@master
+      - name: Update delete items deployment image
+        uses: steebchen/kubectl@v2.0.0
+        with:
+          config: ${{ secrets.KUBE_CONFIG_DATA }}
+          version: v1.27.4
+          command: set image --record deployment/delete-files-consumer delete-files-consumer=${{ secrets.DOCKERHUB_USERNAME }}/background-tasks:${{ github.sha }}
+      - name: Verify successful deployment
+        uses: steebchen/kubectl@v2.0.0
+        with:
+          config: ${{ secrets.KUBE_CONFIG_DATA }}
+          version: v1.27.4
+          command: rollout status deployment/delete-files-consumer
+      - name: Update delete file versions deployment image
+        uses: steebchen/kubectl@v2.0.0
+        with:
+          config: ${{ secrets.KUBE_CONFIG_DATA }}
+          version: v1.27.4
+          command: set image --record deployment/delete-file-versions-consumer delete-file-versions-consumer=${{ secrets.DOCKERHUB_USERNAME }}/background-tasks:${{ github.sha }}
+      - name: Verify successful deployment
+        uses: steebchen/kubectl@v2.0.0
+        with:
+          config: ${{ secrets.KUBE_CONFIG_DATA }}
+          version: v1.27.4
+          command: rollout status deployment/delete-file-versions-consumer

.github/workflows/build-and-deploy-producer.yaml

@@ -1,7 +1,7 @@
 name: build & deploy
 on:
   push:
-    branches: ["master", "feature/cd"]
+    branches: ["master", "feature/cd", "feat/delete-files"]
 jobs:
   build:
     runs-on: ubuntu-latest
@@ -19,25 +19,37 @@ jobs:
         uses: docker/build-push-action@v5
         with:
           context: ./
-          file: ./producer.Dockerfile
+          file: ./Dockerfile
           push: true
-          tags: ${{ secrets.DOCKERHUB_USERNAME }}/drive-background-tasks-producer:${{ github.sha }}
+          tags: ${{ secrets.DOCKERHUB_USERNAME }}/background-tasks:${{ github.sha }}
   deploy:
     needs: build
     runs-on: ubuntu-latest
     environment:
       name: production
     steps:
-      - uses: actions/checkout@master
-      - name: Update deployment image
+      # - uses: actions/checkout@master
+      # - name: Update delete files/folders deployment image
+      #  uses: steebchen/kubectl@v2.0.0
+      #  with:
+      #    config: ${{ secrets.KUBE_CONFIG_DATA }}
+      #    version: v1.27.4
+      #    command: set image --record deployment/background-tasks-producer background-tasks-producer=${{ secrets.DOCKERHUB_USERNAME }}/drive-background-tasks:${{ github.sha }}
+      # - name: Verify succesful deployment
+      #  uses: steebchen/kubectl@v2.0.0
+      #  with:
+      #    config: ${{ secrets.KUBE_CONFIG_DATA }}
+      #    version: v1.27.4
+      #    command: rollout status deployment/background-tasks-producer
+      - name: Update delete file versions deployment image
         uses: steebchen/kubectl@v2.0.0
         with:
           config: ${{ secrets.KUBE_CONFIG_DATA }}
           version: v1.27.4
-          command: set image --record deployment/background-tasks-producer background-tasks-producer=${{ secrets.DOCKERHUB_USERNAME }}/drive-background-tasks-producer:${{ github.sha }}
+          command: set image --record deployment/delete-file-versions-producer delete-file-versions-producer=${{ secrets.DOCKERHUB_USERNAME }}/background-tasks:${{ github.sha }}
       - name: Verify succesful deployment
         uses: steebchen/kubectl@v2.0.0
         with:
           config: ${{ secrets.KUBE_CONFIG_DATA }}
           version: v1.27.4
-          command: rollout status deployment/background-tasks-producer
+          command: rollout status deployment/delete-file-versions-producer

producer.Dockerfile → Dockerfile

@@ -9,4 +9,4 @@ COPY . .
 RUN yarn && yarn build && yarn --production && yarn cache clean
 
 # Start server
-CMD yarn start:prod:producer
+CMD yarn start:prod

consumer.Dockerfile

@@ -1,4 +1,4 @@
-FROM node:iron-slim
+FROM node:24
 LABEL author="internxt"
 
 WORKDIR /app
@@ -9,4 +9,4 @@ COPY . .
 RUN yarn && yarn build && yarn --production && yarn cache clean
 
 # Start server
-CMD yarn start:prod:consumer
+CMD yarn start:prod:consumer

index.ts

@@ -3,50 +3,34 @@ import amqp from 'amqplib';
 import { v4 } from 'uuid';
 
 import { createLogger } from './src/utils';
-import { Consumer } from './src/consumer';
-import { Producer } from './src/producer';
 import { DriveDatabase } from './src/drive';
-import { DeletedFoldersIterator } from './src/tasks/process-folder-deletion/deleted-folders.iterator';
+import { taskTypes, tasks } from './src/tasks';
+import { ProcessType } from './src/tasks/process';
 
-const [,, ...args] = process.argv;
-const [type] = args;
+config();
 
-if (!type) {
-  console.error('Missing argument: type');
-  process.exit(1);
-}
+const taskType = process.env.TASK_TYPE as undefined | string;
 
-if (type !== 'producer' && type !== 'consumer') {
-  console.error('Invalid argument: type. Accepted values are "producer" or "consumer"');
+if (!taskType || !taskTypes.includes(taskType)) {
+  console.error(`Invalid or missing task type. Expected ${
+    taskTypes.map(t => `'${t}'`).join(', ')
+  } but got '${taskType}'`);
   process.exit(1);
 }
-const processId = v4();
-const logger = createLogger(processId);
 
-config();
-
-const amqpServer = process.env.AMQP_SERVER;
-const queueName = process.env.MARK_DELETED_ITEMS_QUEUE_NAME;
-const maxEnqueuedItems = process.env.TASK_MARK_DELETED_ITEMS_PRODUCER_MAX_ENQUEUED_ITEMS;
-const maxConcurrentItems = process.env.TASK_MARK_DELETED_ITEMS_CONSUMER_MAX_CONCURRENT_ITEMS;
+const processType = process.env.PROCESS_TYPE as undefined | ProcessType;
 
-if (!maxEnqueuedItems) {
-  logger.log('Missing env var: TASK_MARK_DELETED_ITEMS_PRODUCER_MAX_ENQUEUED_ITEMS');
+if (!processType || !Object.values(ProcessType).includes(processType)) {
+  console.error(`Invalid or missing process type. Expected ${
+    Object.values(ProcessType).map(t => `'${t}'`).join(', ')
+  } but got '${processType}'`);
   process.exit(1);
 }
 
-if (!maxConcurrentItems) {
-  logger.log('Missing env var: TASK_MARK_DELETED_ITEMS_CONSUMER_MAX_CONCURRENT_ITEMS');
-  process.exit(1);
-}
+const processId = v4();
+const logger = createLogger(processId);
 
-logger.log(`params: process_type -> ${type}, env -> ${
-  JSON.stringify({ 
-    maxConcurrentItems, 
-    maxEnqueuedItems, 
-    queueName 
-  })
-}`);
+const amqpServer = process.env.QUEUE_SERVER;
 
 let db: DriveDatabase;
 let connection: amqp.Connection;
@@ -75,7 +59,7 @@ function handleStop() {
   }
 }
 
-async function start(): Promise<{ connection: amqp.Connection, db: DriveDatabase }> {
+async function start(): Promise<void> {
   db = new DriveDatabase();
 
   logger.log('(drive-db) connecting ...');
@@ -86,63 +70,13 @@ async function start(): Promise<{ connection: amqp.Connection, db: DriveDatabase
   connection = await amqp.connect(amqpServer as string);
   logger.log('(rabbit) connected !');
 
-  return { connection, db };
+  await tasks[taskType as string](processType as 'consumer' | 'producer', { db }, connection);
 }
 
-start().then(({ connection, db }) => {
-  if (type === 'producer') {
-    const deletedFoldersIterator = new DeletedFoldersIterator(db);
-
-    return connection.createChannel().then((channel) => {
-      const producer = new Producer(
-        channel, 
-        queueName as string, 
-        deletedFoldersIterator,
-        maxEnqueuedItems ? parseInt(maxEnqueuedItems as string) : undefined,
-      );
-
-      producer.on('enqueue', (item) => {
-        logger.log(`enqueued item: + ${JSON.stringify(item)}`, 'producer');
-      });
-
-      producer.on('queue-full', () => {
-        logger.log(`queue full, waiting 1s...`, 'producer');
-      });
-
-      return producer.run();
-    });
-  } else {
-    connection.createChannel().then((channel) => {
-      const consumer = new Consumer<{ 
-        folder_id: string,
-        processed: boolean,
-        created_at: Date,
-        updated_at: Date,
-        processed_at: Date,
-      }>(
-        channel, 
-        queueName as string,
-        async (taskPayload) => {
-          logger.log(`received item: + ${JSON.stringify(taskPayload)}`, 'consumer');
-
-          await db.markChildrenFilesAsDeleted(taskPayload.folder_id);
-          await db.markChildrenFoldersAsDeleted(taskPayload.folder_id);
-          await db.markDeletedFolderAsProcessed([taskPayload.folder_id]);
-        },
-        maxConcurrentItems ? parseInt(maxConcurrentItems as string) : undefined,
-      );
-
-      consumer.on('error', ({ err, msg }) => {
-        logger.error(`error processing item: ${JSON.stringify(msg.content)}`, err, 'consumer');
-      });
-
-      consumer.run();
-    });
-  }
-}).catch((err) => {
+start().catch((err) => {
   logger.error('Error starting', err);
   process.exit(1);
-})
+});
 
 process.on('uncaughtException', (err) => {
   logger.error('Uncaught exception', err);

package.json

@@ -4,22 +4,23 @@
   "main": "index.js",
   "license": "MIT",
   "scripts": {
-    "start:dev:producer": "ts-node index.ts producer",
-    "start:dev:consumer": "ts-node index.ts consumer",
-    "start:prod:producer": "node dist/index.js producer",
-    "start:prod:consumer": "node dist/index.js consumer",
+    "start:dev": "ts-node index.ts",
+    "start:prod": "node dist/index.js",
     "build": "tsc"
   },
   "dependencies": {
     "amqplib": "^0.10.3",
+    "axios": "^1.6.1",
     "dotenv": "^16.3.1",
+    "jsonwebtoken": "^9.0.2",
     "pg": "^8.11.3",
     "ts-node": "^10.9.1",
     "typescript": "^5.2.2",
     "uuid": "^9.0.1"
   },
   "devDependencies": {
     "@types/amqplib": "^0.10.3",
+    "@types/jsonwebtoken": "^9.0.5",
     "@types/node": "^20.8.9",
     "@types/pg": "^8.10.7",
     "@types/uuid": "^9.0.6"