Add media storage (#25)

* Adds basic structure for remote storage

* Gets image uploading to a working state

* Adds validator in front of image upload

* Adds delete method for s3 files

* Allow for updating an image

Force name setting

* Fixes test cases

.github/workflows/test.yml

@@ -75,3 +75,4 @@ jobs:
         DB_PORT: 3306
         APP_BASE_URL: https://parcelvoy.com
         QUEUE_DRIVER: memory
+        STORAGE_DRIVER: s3

db/migrations/20220920022913_add_media_library.js

@@ -0,0 +1,24 @@
+exports.up = function(knex) {
+    return knex.schema
+        .createTable('images', function(table) {
+            table.increments()
+            table.integer('project_id')
+                .unsigned()
+                .notNullable()
+                .references('id')
+                .inTable('projects')
+                .onDelete('CASCADE')
+            table.string('uuid', 255).notNullable()
+            table.string('name', 255).defaultTo('')
+            table.string('original_name')
+            table.string('extension')
+            table.string('alt')
+            table.integer('file_size')
+            table.timestamp('created_at').defaultTo(knex.fn.now())
+            table.timestamp('updated_at').defaultTo(knex.fn.now())
+        })
+}
+
+exports.down = function(knex) {
+    return knex.schema.dropTable('images')
+}

package.json

@@ -12,10 +12,12 @@
   "author": "",
   "license": "ISC",
   "devDependencies": {
+    "@types/busboy": "^1.5.0",
     "@types/jest": "^28.1.6",
     "@types/jsonpath": "^0.2.0",
     "@types/jsonwebtoken": "^8.5.9",
     "@types/koa__router": "^8.0.11",
+    "@types/node": "^18.7.18",
     "@types/node-pushnotifications": "^1.0.4",
     "@types/node-schedule": "^2.1.0",
     "@types/nodemailer": "^6.4.4",
@@ -31,14 +33,17 @@
   },
   "dependencies": {
     "@apideck/better-ajv-errors": "^0.3.6",
+    "@aws-sdk/client-s3": "^3.171.0",
     "@aws-sdk/client-ses": "^3.121.0",
     "@aws-sdk/client-sns": "^3.121.0",
-    "@aws-sdk/client-sqs": "^3.121.0",
+    "@aws-sdk/client-sqs": "^3.171.0",
+    "@aws-sdk/lib-storage": "^3.171.0",
     "@koa/cors": "^3.3.0",
     "@koa/router": "^11.0.1",
     "@rxfork/sqs-consumer": "^6.0.0",
     "@types/koa__cors": "^3.3.0",
     "ajv": "^8.11.0",
+    "busboy": "^1.6.0",
     "date-fns": "^2.29.2",
     "date-fns-tz": "^1.3.7",
     "dotenv": "^16.0.1",

src/api.ts

@@ -24,7 +24,7 @@ export default class Api extends Koa {
                 await next()
             } catch (err) {
                 if (err instanceof RequestError) {
-                    return ctx.throw(err.status, err.message)
+                    return ctx.throw(err.message, err.statusCode)
                 }
                 throw err
             }

src/app.ts

@@ -1,9 +1,11 @@
 import Api from './api'
 import loadDatabase, { Database, migrate } from './config/database'
 import loadQueue from './config/queue'
+import loadStorage from './config/storage'
 import { Env } from './config/env'
 import scheduler from './config/scheduler'
 import Queue from './queue'
+import Storage from './storage'
 
 export default class App {
     private static $main: App
@@ -24,8 +26,11 @@ export default class App {
         // Load queue
         const queue = loadQueue(env.queue)
 
+        // Load storage
+        const storage = loadStorage(env.storage)
+
         // Setup app
-        App.$main = new App(env, database, queue)
+        App.$main = new App(env, database, queue, storage)
 
         return App.$main
     }
@@ -39,6 +44,7 @@ export default class App {
         public env: Env,
         public db: Database,
         public queue: Queue,
+        public storage: Storage,
     ) {
         this.api = new Api(this)
         this.scheduler = scheduler(this)

src/config/controllers.ts

@@ -7,6 +7,7 @@ import CampaignController from '../campaigns/CampaignController'
 import ListController from '../lists/ListController'
 import SubscriptionController from '../subscriptions/SubscriptionController'
 import JourneyController from '../journey/JourneyController'
+import ImageController from '../storage/ImageController'
 
 export default (api: import('../api').default) => {
 
@@ -33,6 +34,8 @@ export default (api: import('../api').default) => {
 
     admin.use(JourneyController.routes()).use(JourneyController.allowedMethods())
 
+    admin.use(ImageController.routes()).use(ImageController.allowedMethods())
+
     api.use(admin.routes()).use(admin.allowedMethods())
 
     api.use(client.routes()).use(client.allowedMethods())

src/config/env.ts

@@ -1,9 +1,12 @@
 import * as dotenv from 'dotenv'
+import { StorageConfig } from '../storage/Storage'
+import { QueueConfig } from '../queue/Queue'
 import { DatabaseConfig } from './database'
 
 export interface Env {
     db: DatabaseConfig
     queue: QueueConfig
+    storage: StorageConfig
     port: number
     secret: string
 }
@@ -44,6 +47,16 @@ export default (type?: EnvType): Env => {
                 },
             }),
         }),
+        storage: driver<StorageConfig>(process.env.STORAGE_DRIVER, {
+            s3: () => ({
+                bucket: process.env.AWS_S3_BUCKET!,
+                region: process.env.AWS_REGION!,
+                credentials: {
+                    accessKeyId: process.env.AWS_ACCESS_KEY_ID!,
+                    secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY!,
+                },
+            }),
+        }),
         port: parseInt(process.env.PORT!),
         secret: process.env.APP_SECRET!,
     }

src/config/storage.ts

@@ -0,0 +1,5 @@
+import Storage, { StorageConfig } from '../storage/Storage'
+
+export default (config: StorageConfig) => {
+    return new Storage(config)
+}

src/core/Model.ts

@@ -1,6 +1,6 @@
 import App from '../app'
 import { Database } from '../config/database'
-import { pascalToSnakeCase, pluralize } from '../utilities'
+import { snakeCase, pluralize } from '../utilities'
 
 export const raw = (raw: Database.Value, db: Database = App.main.db) => {
     return db.raw(raw)
@@ -146,7 +146,7 @@ export default class Model {
     }
 
     static get tableName(): string {
-        return pluralize(pascalToSnakeCase(this.name))
+        return pluralize(snakeCase(this.name))
     }
 
     static table(db: Database = App.main.db): Database.QueryBuilder<any> {

src/core/errors.ts

@@ -1,11 +1,30 @@
-
+export interface ErrorType {
-export class RequestError extends Error {
+    message: string
-
+    code: number
-    constructor(
+    statusCode?: number
-        message: string,
-        public readonly status: number,
-    ) {
-        super(message)
-    }
-
 }
+
+export class InternalError extends Error {
+
+    readonly errorCode?: number
+    readonly statusCode?: number
+    constructor(error: ErrorType)
+    constructor(message: string, statusCode?: number, errorCode?: number)
+    constructor(
+        message: string | ErrorType,
+        statusCode?: number,
+        errorCode?: number,
+    ) {
+        if (typeof message === 'string') {
+            super(message)
+            this.statusCode = statusCode
+            this.errorCode = errorCode
+        } else {
+            super(message.message)
+            this.statusCode = message.statusCode
+            this.errorCode = message.code
+        }
+    }
+}
+
+export class RequestError extends InternalError { }

src/journey/JourneyStep.ts

@@ -6,7 +6,7 @@ import { check } from '../rules/RuleEngine'
 import { getJourneyStep, getUserJourneyStep } from './JourneyRepository'
 import { UserEvent } from '../users/UserEvent'
 import { getCampaign, sendCampaign } from '../campaigns/CampaignService'
-import { pascalToSnakeCase } from '../utilities'
+import { snakeCase } from '../utilities'
 
 export class JourneyUserStep extends Model {
     user_id!: number
@@ -30,7 +30,7 @@ export class JourneyStep extends Model {
     static tableName = 'journey_steps'
     static jsonAttributes = ['data']
 
-    static get type() { return pascalToSnakeCase(this.name) }
+    static get type() { return snakeCase(this.name) }
 
     async step(user: User, type: string) {
         await JourneyUserStep.insert({

src/render/Helpers/String.ts

@@ -1,3 +1,4 @@
+import { snakeCase } from '../../utilities'
 import { isNumber } from './Number'
 import { checkType, isType } from './Util'
 
@@ -129,9 +130,7 @@ export const reverse = function(str: string): string {
  */
 export const snakecase = function(str: string): string {
     if (!isString(str)) return ''
-    return str.replace(/[A-Z]/g, (letter, index) => {
+    return snakeCase(str)
-        return index === 0 ? letter.toLowerCase() : '_' + letter.toLowerCase()
-    })
 }
 
 /**

src/storage/Image.ts

@@ -0,0 +1,32 @@
+import Model from '../core/Model'
+import { combineURLs } from '../utilities'
+
+export default class Image extends Model {
+    project_id!: number
+    uuid!: string
+    name!: string
+    original_name!: string
+    extension!: string
+    alt!: string
+    file_size!: number
+
+    get filename(): string {
+        return `${this.uuid}${this.extension}`
+    }
+
+    get url(): string {
+        return combineURLs([process.env.STORAGE_BASE_URL!, this.filename])
+    }
+
+    toJSON() {
+        return {
+            ...this,
+            url: this.url,
+        }
+    }
+}
+
+export interface ImageParams {
+    name: string
+    alt?: string
+}

src/storage/ImageController.ts

@@ -0,0 +1,85 @@
+import Router from '@koa/router'
+import type App from '../app'
+import { JSONSchemaType, validate } from '../core/validate'
+import parse, { ImageMetadata } from './ImageStream'
+import { allImages, getImage, updateImage, uploadImage } from './ImageService'
+import Image, { ImageParams } from './Image'
+
+const router = new Router<{
+    app: App
+    image?: Image
+    user: { project_id: number }
+}>({
+    prefix: '/images',
+})
+
+const uploadMetadata: JSONSchemaType<ImageMetadata> = {
+    $id: 'uploadMetadata',
+    type: 'object',
+    required: ['fieldName', 'fileName', 'mimeType'],
+    properties: {
+        fieldName: {
+            type: 'string',
+        },
+        fileName: {
+            type: 'string',
+        },
+        mimeType: {
+            type: 'string',
+            enum: ['image/jpeg', 'image/gif', 'image/png', 'image/jpg'],
+        },
+        size: {
+            type: 'number',
+        },
+    },
+    additionalProperties: false,
+}
+
+router.post('/', async ctx => {
+    const stream = await parse(ctx)
+
+    // Validate but we don't need the response since we already have it
+    validate(uploadMetadata, stream.metadata)
+
+    ctx.body = await uploadImage(ctx.state.user.project_id, stream)
+})
+
+router.get('/', async ctx => {
+    ctx.body = await allImages(ctx.state.user.project_id)
+})
+
+router.param('imageId', async (value, ctx, next) => {
+    ctx.state.image = await getImage(parseInt(ctx.params.imageId), ctx.state.user.project_id)
+    if (!ctx.state.image) {
+        ctx.throw(404)
+        return
+    }
+    return await next()
+})
+
+router.get('/:imageId', async ctx => {
+    ctx.body = ctx.state.image
+})
+
+const imageUpdateMetadata: JSONSchemaType<ImageParams> = {
+    $id: 'imageUpdateMetadata',
+    type: 'object',
+    required: ['name'],
+    properties: {
+        name: {
+            type: 'string',
+        },
+        alt: {
+            type: 'string',
+            nullable: true,
+        },
+    },
+    additionalProperties: false,
+}
+
+router.patch('/:imageId', async ctx => {
+    const payload = validate(imageUpdateMetadata, ctx.request.body)
+    ctx.body = await updateImage(ctx.state.image!.id, payload)
+})
+
+export default router

src/storage/ImageService.ts

@@ -0,0 +1,25 @@
+import App from '../app'
+import { snakeCase } from '../utilities'
+import Image, { ImageParams } from './Image'
+import { ImageStream } from './ImageStream'
+
+export const uploadImage = async (projectId: number, stream: ImageStream): Promise<Image> => {
+    const upload = await App.main.storage.upload(stream)
+    return await Image.insertAndFetch({
+        project_id: projectId,
+        name: upload.original_name ? snakeCase(upload.original_name) : '',
+        ...upload,
+    })
+}
+
+export const allImages = async (projectId: number): Promise<Image[]> => {
+    return await Image.all(qb => qb.where('project_id', projectId))
+}
+
+export const getImage = async (projectId: number, id: number): Promise<Image | undefined> => {
+    return await Image.find(id, qb => qb.where('project_id', projectId))
+}
+
+export const updateImage = async (id: number, params: ImageParams): Promise<Image | undefined> => {
+    return await Image.updateAndFetch(id, params)
+}

src/storage/ImageStream.ts

@@ -0,0 +1,67 @@
+import { Context } from 'koa'
+import Busboy from 'busboy'
+import { Stream } from 'stream'
+import { RequestError } from '../core/errors'
+import StorageError from './StorageError'
+
+export interface ImageMetadata {
+    fieldName: string
+    fileName: string
+    mimeType: string
+    size: number
+}
+
+export interface ImageStream {
+    file: Stream
+    metadata: ImageMetadata
+}
+
+export default function parse(ctx: Context): Promise<ImageStream> {
+    return new Promise((resolve, reject) => {
+        if (!ctx.is('multipart')) {
+            reject(new RequestError(StorageError.BadFormType))
+            return
+        }
+
+        const busboy = Busboy({
+            headers: ctx.req.headers,
+            limits: {
+                files: 1, // Allow only a single upload at a time.
+            },
+        })
+
+        busboy.once('file', onFile)
+        busboy.once('error', onError)
+        busboy.once('close', onClose)
+        ctx.req.pipe(busboy)
+
+        function cleanup() {
+            busboy.removeListener('file', onFile)
+            busboy.removeListener('error', onError)
+            busboy.removeListener('close', onClose)
+        }
+
+        function onFile(fieldName: string, file: Stream, info: { filename: string, mimeType: string }) {
+            cleanup()
+            resolve({
+                file,
+                metadata: {
+                    fieldName,
+                    fileName: info.filename,
+                    mimeType: info.mimeType,
+                    size: parseInt(ctx.req.headers['content-length'] ?? '0'),
+                },
+            })
+        }
+
+        function onError(error: Error) {
+            cleanup()
+            reject(error)
+        }
+
+        function onClose() {
+            cleanup()
+            reject(new RequestError(StorageError.NoFilesUploaded))
+        }
+    })
+}

src/storage/S3StorageProvider.ts

@@ -0,0 +1,45 @@
+import { S3 } from '@aws-sdk/client-s3'
+import { Upload } from '@aws-sdk/lib-storage'
+import { PassThrough } from 'stream'
+import { AWSConfig } from '../core/aws'
+import { StorageTypeConfig } from './Storage'
+import { ImageUploadTask, StorageProvider } from './StorageProvider'
+
+export interface S3Config extends StorageTypeConfig, AWSConfig {
+    bucket: string
+}
+
+export class S3StorageProvider implements StorageProvider {
+
+    config: S3Config
+
+    constructor(config: S3Config) {
+        this.config = config
+    }
+
+    async upload(task: ImageUploadTask) {
+        const pass = new PassThrough()
+        const s3 = new S3(this.config)
+
+        const promise = new Upload({
+            client: s3,
+            params: {
+                Key: task.url,
+                Body: pass,
+                Bucket: this.config.bucket,
+            },
+        }).done()
+
+        task.stream.pipe(pass)
+
+        await promise
+    }
+
+    async delete(filename: string): Promise<void> {
+        const s3 = new S3(this.config)
+        await s3.deleteObject({
+            Bucket: this.config.bucket,
+            Key: filename,
+        })
+    }
+}

src/storage/Storage.ts

@@ -0,0 +1,50 @@
+import { DriverConfig } from '../config/env'
+import { ImageStream } from './ImageStream'
+import Image from './Image'
+import { S3Config, S3StorageProvider } from './S3StorageProvider'
+import { StorageProvider, StorageProviderName } from './StorageProvider'
+import path from 'path'
+import { uuid } from '../utilities'
+import { InternalError } from '../core/errors'
+import StorageError from './StorageError'
+
+export type StorageConfig = S3Config
+export interface StorageTypeConfig extends DriverConfig {
+    driver: StorageProviderName
+}
+
+export interface ImageUpload {
+    extension: string
+}
+
+export default class Storage {
+    provider: StorageProvider
+
+    constructor(config?: StorageConfig) {
+        if (config?.driver === 's3') {
+            this.provider = new S3StorageProvider(config)
+        } else {
+            throw new InternalError(StorageError.UndefinedStorageMethod)
+        }
+    }
+
+    async upload(image: ImageStream): Promise<Partial<Image>> {
+        const key = uuid()
+        const originalPath = path.parse(image.metadata.fileName)
+        const extension = originalPath.ext
+        const fileName = originalPath.name
+        const url = `${key}${extension}`
+
+        await this.provider.upload({
+            stream: image.file,
+            url,
+        })
+
+        return {
+            uuid: key,
+            original_name: fileName,
+            extension,
+            file_size: image.metadata.size,
+        }
+    }
+}

src/storage/StorageError.ts

@@ -0,0 +1,14 @@
+export default {
+    UndefinedStorageMethod: {
+        message: 'A valid storage method must be defined!',
+        code: 5000,
+    },
+    NoFilesUploaded: {
+        message: 'The request contains no files. Please attach a file to upload.',
+        code: 5001,
+    },
+    BadFormType: {
+        message: 'Incorrect form type. Please make sure file is being submitted in a multipart form.',
+        code: 5002,
+    },
+}

src/storage/StorageProvider.ts

@@ -0,0 +1,13 @@
+import { Stream } from 'stream'
+
+export type StorageProviderName = 's3'
+
+export interface ImageUploadTask {
+    stream: Stream
+    url: string
+}
+
+export interface StorageProvider {
+    upload(task: ImageUploadTask): Promise<void>
+    delete(filename: string): Promise<void>
+}

src/storage/index.ts

@@ -0,0 +1,5 @@
+import Image from './Image'
+import Storage from './Storage'
+
+export { Image }
+export default Storage

src/utilities/index.ts

@@ -10,7 +10,9 @@ export const randomInt = (min = 0, max = 100): number => {
     return Math.floor(Math.random() * (max - min + 1)) + min
 }
 
-export const pascalToSnakeCase = (str: string): string => str.split(/(?=[A-Z])/).join('_').toLowerCase()
+export const snakeCase = (str: string): string => str.match(/[A-Z]{2,}(?=[A-Z][a-z]+[0-9]*|\b)|[A-Z]?[a-z]+[0-9]*|[A-Z]|[0-9]+/g)
+    ?.map(x => x.toLowerCase())
+    .join('_') ?? ''
 
 export const uuid = (): string => {
     return crypto.randomUUID()