99 data product uploads

api/config/default.json

@@ -28,6 +28,11 @@
       "client_id": "",
       "client_secret": "",
       "scope_prefix": "download_file:"
+    },
+    "upload": {
+      "client_id": "",
+      "client_secret": "",
+      "scope_prefix": "upload"
     }
   },
   "download_server": {
@@ -82,7 +87,30 @@
           "task": "delete_dataset"
         }
       ]
+    },
+    "process_uploads": {
+      "steps": [
+        {
+          "name": "chunks_to_files",
+          "task": "chunks_to_files"
+        }
+      ]
     }
   },
-  "dataset_types" : ["RAW_DATA", "DATA_PRODUCT"]
-}
+  "dataset_types" : ["RAW_DATA", "DATA_PRODUCT"],
+  "DONE_STATUSES": {
+    "REVOKED": "REVOKED",
+    "FAILURE": "FAILURE",
+    "SUCCESS": "SUCCESS"
+  },
+  "upload_status": {
+    "UPLOADING": "UPLOADING",
+    "UPLOAD_FAILED": "UPLOAD_FAILED",
+    "UPLOADED": "UPLOADED",
+    "PROCESSING": "PROCESSING",
+    "COMPLETE": "COMPLETE",
+    "FAILED": "FAILED",
+    "VALIDATION_FAILED": "VALIDATION_FAILED"
+  },
+  "upload_path": "/opt/sca/data/uploads"
+}

api/package.json

@@ -45,6 +45,7 @@
     "lodash": "^4.17.21",
     "morgan": "^1.10.0",
     "multer": "1.4.5-lts.1",
+    "spark-md5": "^3.0.2",
     "swagger-autogen": "^2.23.5",
     "swagger-ui-express": "^5.0.0",
     "winston": "^3.10.0"

api/prisma/migrations/20231129205028_99_uploads/migration.sql

@@ -0,0 +1,58 @@
+-- CreateEnum
+CREATE TYPE "upload_status" AS ENUM ('UPLOADING', 'UPLOAD_FAILED', 'UPLOADED', 'PROCESSING', 'COMPLETE', 'VALIDATION_FAILED', 'FAILED');
+
+-- AlterTable
+ALTER TABLE "dataset" ADD COLUMN     "file_type_id" INTEGER;
+
+-- CreateTable
+CREATE TABLE "upload_log" (
+    "id" SERIAL NOT NULL,
+    "status" "upload_status" NOT NULL,
+    "dataset_id" INTEGER,
+    "last_updated" TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "processing_attempt_count" INTEGER NOT NULL DEFAULT 0,
+    "user_id" INTEGER NOT NULL,
+
+    CONSTRAINT "upload_log_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "file_upload_log" (
+    "id" SERIAL NOT NULL,
+    "name" TEXT NOT NULL,
+    "md5" TEXT NOT NULL,
+    "num_chunks" INTEGER NOT NULL,
+    "status" "upload_status" NOT NULL,
+    "chunks_path" TEXT NOT NULL,
+    "destination_path" TEXT NOT NULL,
+    "upload_log_id" INTEGER,
+
+    CONSTRAINT "file_upload_log_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "dataset_file_type" (
+    "id" SERIAL NOT NULL,
+    "name" TEXT NOT NULL,
+    "extension" TEXT NOT NULL,
+
+    CONSTRAINT "dataset_file_type_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "upload_log_dataset_id_key" ON "upload_log"("dataset_id");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "dataset_file_type_name_extension_key" ON "dataset_file_type"("name", "extension");
+
+-- AddForeignKey
+ALTER TABLE "dataset" ADD CONSTRAINT "dataset_file_type_id_fkey" FOREIGN KEY ("file_type_id") REFERENCES "dataset_file_type"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "upload_log" ADD CONSTRAINT "upload_log_dataset_id_fkey" FOREIGN KEY ("dataset_id") REFERENCES "dataset"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "upload_log" ADD CONSTRAINT "upload_log_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "user"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "file_upload_log" ADD CONSTRAINT "file_upload_log_upload_log_id_fkey" FOREIGN KEY ("upload_log_id") REFERENCES "upload_log"("id") ON DELETE SET NULL ON UPDATE CASCADE;

api/prisma/schema.prisma

@@ -11,6 +11,8 @@ model dataset {
   id               Int                 @id @default(autoincrement())
   name             String
   type             String
+  file_type        dataset_file_type?  @relation(fields: [file_type_id], references: [id])
+  file_type_id     Int?
   num_directories  Int?
   num_files        Int?
   du_size          BigInt?
@@ -34,9 +36,44 @@ model dataset {
   projects         project_dataset[]
   accesses         data_access_log[]
   stage_requests   stage_request_log[]
+  upload_log       upload_log?
+
   @@unique([name, type, is_deleted])
 }
 
+model upload_log {
+  id                       Int               @id @default(autoincrement())
+  status                   upload_status
+  dataset_id               Int?              @unique
+  dataset                  dataset?          @relation(fields: [dataset_id], references: [id])
+  files                    file_upload_log[]
+  last_updated             DateTime          @default(now()) @db.Timestamp(6)
+  processing_attempt_count Int               @default(0)
+  user_id                  Int
+  user                     user              @relation(fields: [user_id], references: [id])
+}
+
+model file_upload_log {
+  id               Int           @id @default(autoincrement())
+  name             String
+  md5              String
+  num_chunks       Int
+  status           upload_status
+  chunks_path      String
+  destination_path String
+  upload_log       upload_log?   @relation(fields: [upload_log_id], references: [id])
+  upload_log_id    Int?
+}
+
+model dataset_file_type {
+  id        Int       @id @default(autoincrement())
+  name      String
+  extension String
+  datasets  dataset[]
+
+  @@unique([name, extension])
+}
+
 model dataset_hierarchy {
   source_id       Int
   derived_id      Int
@@ -48,19 +85,20 @@ model dataset_hierarchy {
 }
 
 model dataset_file {
-  id             Int                      @id @default(autoincrement())
-  name           String?
-  path           String
-  md5            String?
-  size           BigInt?
-  filetype       String?
-  metadata       Json?
-  status         String?
-  dataset_id     Int
-  dataset        dataset                  @relation(fields: [dataset_id], references: [id], onDelete: Cascade)
-  parents        dataset_file_hierarchy[] @relation("child")
-  children       dataset_file_hierarchy[] @relation("parent")
-  accesses       data_access_log[]
+  id         Int                      @id @default(autoincrement())
+  name       String?
+  path       String
+  md5        String?
+  size       BigInt?
+  filetype   String?
+  metadata   Json?
+  status     String?
+  dataset_id Int
+  dataset    dataset                  @relation(fields: [dataset_id], references: [id], onDelete: Cascade)
+  parents    dataset_file_hierarchy[] @relation("child")
+  children   dataset_file_hierarchy[] @relation("parent")
+  accesses   data_access_log[]
+
   @@unique([path, dataset_id])
   @@index([dataset_id])
 }
@@ -97,36 +135,36 @@ model dataset_state {
 }
 
 model data_access_log {
-  id            Int           @id @default(autoincrement())
-  timestamp     DateTime      @default(now()) @db.Timestamp(6)
-  access_type   access_type
-  file_id       Int?
-  dataset_file  dataset_file? @relation(fields: [file_id], references: [id])
-  dataset_id    Int?
-  dataset       dataset?      @relation(fields: [dataset_id], references: [id])
-  user_id       Int
-  user          user          @relation(fields: [user_id], references: [id])
+  id           Int           @id @default(autoincrement())
+  timestamp    DateTime      @default(now()) @db.Timestamp(6)
+  access_type  access_type
+  file_id      Int?
+  dataset_file dataset_file? @relation(fields: [file_id], references: [id])
+  dataset_id   Int?
+  dataset      dataset?      @relation(fields: [dataset_id], references: [id])
+  user_id      Int
+  user         user          @relation(fields: [user_id], references: [id])
 }
 
 model stage_request_log {
-  id            Int           @id @default(autoincrement())
-  timestamp     DateTime      @default(now()) @db.Timestamp(6)
-  dataset_id    Int?
-  dataset       dataset?      @relation(fields: [dataset_id], references: [id])
-  user_id       Int
-  user          user          @relation(fields: [user_id], references: [id])
+  id         Int      @id @default(autoincrement())
+  timestamp  DateTime @default(now()) @db.Timestamp(6)
+  dataset_id Int?
+  dataset    dataset? @relation(fields: [dataset_id], references: [id])
+  user_id    Int
+  user       user     @relation(fields: [user_id], references: [id])
 }
 
 model user {
-  id             Int             @id @default(autoincrement())
-  username       String          @unique @db.VarChar(100)
-  name           String?         @db.VarChar(100)
-  email          String          @unique @db.VarChar(100)
-  cas_id         String?         @unique @db.VarChar(100)
+  id             Int                 @id @default(autoincrement())
+  username       String              @unique @db.VarChar(100)
+  name           String?             @db.VarChar(100)
+  email          String              @unique @db.VarChar(100)
+  cas_id         String?             @unique @db.VarChar(100)
   notes          String?
-  created_at     DateTime        @default(now()) @db.Timestamp(6)
-  updated_at     DateTime        @default(now()) @updatedAt @db.Timestamp(6)
-  is_deleted     Boolean         @default(false)
+  created_at     DateTime            @default(now()) @db.Timestamp(6)
+  updated_at     DateTime            @default(now()) @updatedAt @db.Timestamp(6)
+  is_deleted     Boolean             @default(false)
   user_role      user_role[]
   settings       user_settings?
   contacts       contact[]
@@ -136,6 +174,7 @@ model user {
   login          user_login?
   accesses       data_access_log[]
   stage_requests stage_request_log[]
+  uploads        upload_log[]
 }
 
 model user_password {
@@ -280,4 +319,14 @@ model worker_process {
 enum access_type {
   BROWSER
   SLATE_SCRATCH
-}
+}
+
+enum upload_status {
+  UPLOADING
+  UPLOAD_FAILED
+  UPLOADED
+  PROCESSING
+  COMPLETE
+  VALIDATION_FAILED
+  FAILED
+}

api/prisma/seed.js

@@ -143,6 +143,11 @@ async function main() {
 
   await Promise.all(operator_promises);
 
+  // create dataset_file_type
+  await prisma.dataset_file_type.deleteMany();
+  await prisma.dataset_file_type.createMany({ data: data.dataset_file_types });
+
+  // create datasets
   const datasetPromises = data.datasets.map((dataset) => {
     const { workflows, ...dataset_obj } = dataset;
     if (workflows) {
@@ -281,6 +286,26 @@ async function main() {
   await prisma.stage_request_log.createMany({
     data: stage_request_logs,
   });
+
+  // create data upload entries
+  const dataset_uploads = data.data_product_upload_logs;
+  const operators = await prisma.user.findMany({
+    where: {
+      user_role: {
+        some: {
+          role_id: 2,
+        },
+      },
+    },
+  });
+  const uploads = dataset_uploads.map((e) => {
+    const operator = operators[Math.floor(Math.random() * operators.length)];
+    return { ...e, user_id: operator.id };
+  });
+  await prisma.upload_log.deleteMany();
+  await prisma.upload_log.createMany({
+    data: uploads,
+  });
 }
 
 main()