Rejected uploads handling

cli/.gitignore

@@ -6,3 +6,4 @@
 /tmp
 /yarn.lock
 node_modules
+/examples/content/*__rejectedContent.json

cli/.prettierignore

@@ -1,2 +1,3 @@
 /lib/
 .nyc_output
+/examples

cli/src/Types.ts

@@ -15,7 +15,7 @@ import {
   ChannelCategoryMetadata,
   VideoCategoryMetadata,
 } from '@joystream/content-metadata-protobuf'
-import { ContentParameters } from '@joystream/types/storage'
+import { ContentId, ContentParameters } from '@joystream/types/storage'
 
 // KeyringPair type extended with mandatory "meta.name"
 // It's used for accounts/keys management within CLI.
@@ -210,9 +210,13 @@ export enum AssetType {
   AnyAsset = 1,
 }
 
-export type InputAssetDetails = {
-  parameters: ContentParameters
+export type InputAsset = {
   path: string
+  contentId: ContentId
+}
+
+export type InputAssetDetails = InputAsset & {
+  parameters: ContentParameters
 }
 
 export type VideoFFProbeMetadata = {

cli/src/base/UploadCommandBase.ts

@@ -1,7 +1,8 @@
 import ContentDirectoryCommandBase from './ContentDirectoryCommandBase'
-import { VideoFFProbeMetadata, VideoFileMetadata, AssetType, InputAssetDetails } from '../Types'
+import { VideoFFProbeMetadata, VideoFileMetadata, AssetType, InputAsset, InputAssetDetails } from '../Types'
 import { ContentId, ContentParameters } from '@joystream/types/storage'
 import { MultiBar, Options, SingleBar } from 'cli-progress'
+import { Assets } from '../json-schemas/typings/Assets.schema'
 import ExitCodes from '../ExitCodes'
 import ipfsHash from 'ipfs-only-hash'
 import fs from 'fs'
@@ -10,6 +11,7 @@ import axios, { AxiosRequestConfig } from 'axios'
 import ffprobeInstaller from '@ffprobe-installer/ffprobe'
 import ffmpeg from 'fluent-ffmpeg'
 import path from 'path'
+import chalk from 'chalk'
 
 ffmpeg.setFfprobePath(ffprobeInstaller.path)
 
@@ -130,7 +132,7 @@ export default abstract class UploadCommandBase extends ContentDirectoryCommandB
     return new URL(`asset/v0/${contentId.encode()}`, endpointRoot).toString()
   }
 
-  async getRandomProviderEndpoint(): Promise<string> {
+  async getRandomProviderEndpoint(): Promise<string | null> {
     const endpoints = _.shuffle(await this.getApi().allStorageProviderEndpoints())
     for (const endpoint of endpoints) {
       try {
@@ -143,7 +145,7 @@ export default abstract class UploadCommandBase extends ContentDirectoryCommandB
       }
     }
 
-    this.error('No active storage provider found', { exit: ExitCodes.ActionCurrentlyUnavailable })
+    return null
   }
 
   async generateContentParameters(filePath: string, type: AssetType): Promise<ContentParameters> {
@@ -165,15 +167,22 @@ export default abstract class UploadCommandBase extends ContentDirectoryCommandB
 
     // Return data
     return await Promise.all(
-      paths.map(async (path) => ({
-        path,
-        parameters: await this.generateContentParameters(path, AssetType.AnyAsset),
-      }))
+      paths.map(async (path) => {
+        const parameters = await this.generateContentParameters(path, AssetType.AnyAsset)
+        return {
+          path,
+          contentId: parameters.content_id,
+          parameters,
+        }
+      })
     )
   }
 
   async uploadAsset(contentId: ContentId, filePath: string, endpoint?: string, multiBar?: MultiBar): Promise<void> {
     const providerEndpoint = endpoint || (await this.getRandomProviderEndpoint())
+    if (!providerEndpoint) {
+      this.error('No active provider found!', { exit: ExitCodes.ActionCurrentlyUnavailable })
+    }
     const uploadUrl = this.assetUrl(providerEndpoint, contentId)
     const fileSize = this.getFileSize(filePath)
     const { fileStream, progressBar } = this.createReadStreamWithProgressBar(
@@ -195,7 +204,7 @@ export default abstract class UploadCommandBase extends ContentDirectoryCommandB
       }
       await axios.put(uploadUrl, fileStream, config)
     } catch (e) {
-      multiBar ? multiBar.stop() : progressBar.stop()
+      progressBar.stop()
       const msg = (e.response && e.response.data && e.response.data.message) || e.message || e
       this.error(`Unexpected error when trying to upload a file: ${msg}`, {
         exit: ExitCodes.ExternalInfrastructureError,
@@ -203,10 +212,65 @@ export default abstract class UploadCommandBase extends ContentDirectoryCommandB
     }
   }
 
-  async uploadAssets(assets: InputAssetDetails[]): Promise<void> {
+  async uploadAssets(
+    assets: InputAsset[],
+    inputFilePath: string,
+    outputFilePostfix = '__rejectedContent'
+  ): Promise<void> {
     const endpoint = await this.getRandomProviderEndpoint()
+    if (!endpoint) {
+      this.warn('No storage provider is currently available!')
+      this.handleRejectedUploads(
+        assets,
+        assets.map(() => false),
+        inputFilePath,
+        outputFilePostfix
+      )
+      this.exit(ExitCodes.ActionCurrentlyUnavailable)
+    }
     const multiBar = new MultiBar(this.progressBarOptions)
-    await Promise.all(assets.map((a) => this.uploadAsset(a.parameters.content_id, a.path, endpoint, multiBar)))
+    // Workaround replacement for Promise.allSettled (which is only available in ES2020)
+    const results = await Promise.all(
+      assets.map(async (a) => {
+        try {
+          await this.uploadAsset(a.contentId, a.path, endpoint, multiBar)
+          return true
+        } catch (e) {
+          return false
+        }
+      })
+    )
+    this.handleRejectedUploads(assets, results, inputFilePath, outputFilePostfix)
     multiBar.stop()
   }
+
+  private handleRejectedUploads(
+    assets: InputAsset[],
+    results: boolean[],
+    inputFilePath: string,
+    outputFilePostfix: string
+  ): void {
+    // Try to save rejected contentIds and paths for reupload purposes
+    const rejectedAssetsOutput: Assets = []
+    results.forEach(
+      (r, i) =>
+        r === false && rejectedAssetsOutput.push({ contentId: assets[i].contentId.encode(), path: assets[i].path })
+    )
+    if (rejectedAssetsOutput.length) {
+      this.warn(
+        `Some assets were not uploaded succesfully. Try reuploading them with ${chalk.white('content:reuploadAssets')}!`
+      )
+      console.log(rejectedAssetsOutput)
+      const outputPath = inputFilePath.replace('.json', `${outputFilePostfix}.json`)
+      try {
+        fs.writeFileSync(outputPath, JSON.stringify(rejectedAssetsOutput, null, 4))
+        this.log(`Rejected content ids succesfully saved to: ${chalk.white(outputPath)}!`)
+      } catch (e) {
+        console.error(e)
+        this.warn(
+          `Could not write rejected content output to ${outputPath}. Try copying the output above and creating the file manually!`
+        )
+      }
+    }
+  }
 }

cli/src/commands/content/createChannel.ts

@@ -56,6 +56,6 @@ export default class CreateChannelCommand extends UploadCommandBase {
 
     await this.sendAndFollowNamedTx(account, 'content', 'createChannel', [actor, channelCreationParameters])
 
-    await this.uploadAssets(inputAssets)
+    await this.uploadAssets(inputAssets, input)
   }
 }

cli/src/commands/content/createVideo.ts

@@ -80,7 +80,6 @@ export default class CreateVideoCommand extends UploadCommandBase {
     await this.sendAndFollowNamedTx(account, 'content', 'createVideo', [actor, channelId, videoCreationParameters])
 
     // Upload assets
-    await this.uploadAssets(inputAssets)
-    // TODO: Reupload option if failed?
+    await this.uploadAssets(inputAssets, input)
   }
 }

cli/src/commands/content/reuploadAssets.ts

@@ -0,0 +1,36 @@
+import UploadCommandBase from '../../base/UploadCommandBase'
+import { getInputJson } from '../../helpers/InputOutput'
+import AssetsSchema from '../../json-schemas/Assets.schema.json'
+import { Assets as AssetsInput } from '../../json-schemas/typings/Assets.schema'
+import { flags } from '@oclif/command'
+import { ContentId } from '@joystream/types/storage'
+
+export default class ReuploadVideoAssetsCommand extends UploadCommandBase {
+  static description = 'Allows reuploading assets that were not succesfully uploaded during channel/video creation'
+
+  static flags = {
+    input: flags.string({
+      char: 'i',
+      required: true,
+      description: 'Path to JSON file containing array of assets to reupload (contentIds and paths)',
+    }),
+  }
+
+  async run() {
+    const { input } = this.parse(ReuploadVideoAssetsCommand).flags
+
+    // Get context
+    const account = await this.getRequiredSelectedAccount()
+    await this.requestAccountDecoding(account)
+
+    // Get input from file
+    const inputData = await getInputJson<AssetsInput>(input, AssetsSchema)
+    const inputAssets = inputData.map(({ contentId, path }) => ({
+      contentId: ContentId.decode(this.getTypesRegistry(), contentId),
+      path,
+    }))
+
+    // Upload assets
+    await this.uploadAssets(inputAssets, input, '')
+  }
+}

cli/src/commands/content/updateChannel.ts

@@ -68,6 +68,6 @@ export default class UpdateChannelCommand extends UploadCommandBase {
       channelUpdateParameters,
     ])
 
-    await this.uploadAssets(inputAssets)
+    await this.uploadAssets(inputAssets, input)
   }
 }

cli/src/commands/content/updateVideo.ts

@@ -63,6 +63,6 @@ export default class UpdateVideoCommand extends UploadCommandBase {
 
     await this.sendAndFollowNamedTx(currentAccount, 'content', 'updateVideo', [actor, videoId, videoUpdateParameters])
 
-    await this.uploadAssets(inputAssets)
+    await this.uploadAssets(inputAssets, input)
   }
 }

cli/src/helpers/InputOutput.ts

@@ -1,5 +1,6 @@
 import { flags } from '@oclif/command'
 import { CLIError } from '@oclif/errors'
+import Ajv from 'ajv'
 import ExitCodes from '../ExitCodes'
 import fs from 'fs'
 import path from 'path'
@@ -19,7 +20,7 @@ export const IOFlags = {
   }),
 }
 
-export async function getInputJson<T>(inputPath: string): Promise<T> {
+export async function getInputJson<T>(inputPath: string, schema?: Record<string, unknown>): Promise<T> {
   let content, jsonObj
   try {
     content = fs.readFileSync(inputPath).toString()
@@ -31,10 +32,21 @@ export async function getInputJson<T>(inputPath: string): Promise<T> {
   } catch (e) {
     throw new CLIError(`JSON parsing failed for file: ${inputPath}`, { exit: ExitCodes.InvalidInput })
   }
+  if (schema) {
+    await validateInput(jsonObj, schema)
+  }
 
   return jsonObj as T
 }
 
+export async function validateInput(input: unknown, schema: Record<string, unknown>): Promise<void> {
+  const ajv = new Ajv({ allErrors: true })
+  const valid = ajv.validate(schema, input) as boolean
+  if (!valid) {
+    throw new CLIError(`Input JSON file is not valid: ${ajv.errorsText()}`)
+  }
+}
+
 export function saveOutputJson(outputPath: string | undefined, fileName: string, data: any): void {
   if (outputPath) {
     let outputFilePath = path.join(outputPath, fileName)

cli/src/json-schemas/Assets.schema.json

@@ -0,0 +1,22 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema",
+  "$id": "https://joystream.org/Assets.schema.json",
+  "title": "Assets",
+  "description": "List of assets to upload/reupload",
+  "type": "array",
+  "items": {
+    "type": "object",
+    "required": ["contentId", "path"],
+    "additionalProperties": false,
+    "properties": {
+      "contentId": {
+        "type": "string",
+        "description": "Already existing ContentID"
+      },
+      "path": {
+        "type": "string",
+        "description": "Path to the content file (relative to input json file)"
+      }
+    }
+  }
+}

cli/src/json-schemas/typings/Assets.schema.d.ts

@@ -0,0 +1,20 @@
+/* tslint:disable */
+/**
+ * This file was automatically generated by json-schema-to-typescript.
+ * DO NOT MODIFY IT BY HAND. Instead, modify the source JSONSchema file,
+ * and run json-schema-to-typescript to regenerate this file.
+ */
+
+/**
+ * List of assets to upload/reupload
+ */
+export type Assets = {
+  /**
+   * Already existing ContentID
+   */
+  contentId: string
+  /**
+   * Path to the content file (relative to input json file)
+   */
+  path: string
+}[]