feat: make cache server stateless & native multipart uploads

Author: LouisHaftmann

.env

@@ -9,13 +9,12 @@ STORAGE_FILESYSTEM_PATH=.data/storage/filesystem
 
 # s3
 # STORAGE_DRIVER=s3
-# STORAGE_S3_BUCKET=
-# STORAGE_S3_ENDPOINT=
-# STORAGE_S3_REGION=
-# STORAGE_S3_PORT=
-# STORAGE_S3_USE_SSL=
-# STORAGE_S3_ACCESS_KEY=
-# STORAGE_S3_SECRET_KEY=
+# STORAGE_S3_BUCKET=test
+# STORAGE_S3_ENDPOINT=localhost
+# STORAGE_S3_PORT=9000
+# STORAGE_S3_USE_SSL=false
+# STORAGE_S3_ACCESS_KEY=minioadmin
+# STORAGE_S3_SECRET_KEY=minioadmin
 
 # sqlite
 DB_DRIVER=sqlite

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:20-alpine as builder
+FROM node:22-alpine as builder
 
 ARG BUILD_HASH
 ENV BUILD_HASH=${BUILD_HASH}
@@ -21,7 +21,7 @@ RUN pnpm run build
 
 # --------------------------------------------
 
-FROM node:20-alpine as runner
+FROM node:22-alpine as runner
 
 WORKDIR /app
 

eslint.config.js

@@ -14,6 +14,7 @@ export default eslintConfig({
   .append({
     rules: {
       'compat/compat': 'off',
+      'node/prefer-global/buffer': 'off',
     },
   })
   .append({

lib/db/driver.ts renamed to lib/db/defineDatabaseDriver.ts

@@ -1,3 +1,4 @@
+/* eslint-disable unicorn/filename-case */
 import { formatZodError } from '~/lib/env'
 import { logger } from '~/lib/logger'
 

db-drivers/index.ts renamed to lib/db/drivers/index.ts

@@ -1,7 +1,7 @@
-import { mysqlDriver } from '~/db-drivers/mysql'
-import { postgresDriver } from '~/db-drivers/postgres'
-import { sqliteDriver } from '~/db-drivers/sqlite'
-import type { defineDatabaseDriver } from '~/lib/db/driver'
+import type { defineDatabaseDriver } from '~/lib/db/defineDatabaseDriver'
+import { mysqlDriver } from '~/lib/db/drivers/mysql'
+import { postgresDriver } from '~/lib/db/drivers/postgres'
+import { sqliteDriver } from '~/lib/db/drivers/sqlite'
 
 const databaseDrivers = {
   sqlite: sqliteDriver,

db-drivers/mysql.ts renamed to lib/db/drivers/mysql.ts

@@ -2,7 +2,7 @@ import { MysqlDialect } from 'kysely'
 import { createPool } from 'mysql2'
 import { z } from 'zod'
 
-import { defineDatabaseDriver } from '~/lib/db/driver'
+import { defineDatabaseDriver } from '~/lib/db/defineDatabaseDriver'
 
 export const mysqlDriver = defineDatabaseDriver({
   envSchema: z.object({

db-drivers/postgres.ts renamed to lib/db/drivers/postgres.ts

@@ -2,7 +2,7 @@ import { PostgresDialect } from 'kysely'
 import pg from 'pg'
 import { z } from 'zod'
 
-import { defineDatabaseDriver } from '~/lib/db/driver'
+import { defineDatabaseDriver } from '~/lib/db/defineDatabaseDriver'
 
 export const postgresDriver = defineDatabaseDriver({
   envSchema: z.object({

db-drivers/sqlite.ts renamed to lib/db/drivers/sqlite.ts

@@ -5,7 +5,7 @@ import SQLite from 'better-sqlite3'
 import { SqliteDialect } from 'kysely'
 import { z } from 'zod'
 
-import { defineDatabaseDriver } from '~/lib/db/driver'
+import { defineDatabaseDriver } from '~/lib/db/defineDatabaseDriver'
 
 export const sqliteDriver = defineDatabaseDriver({
   envSchema: z.object({

lib/db/index.ts

@@ -1,25 +1,39 @@
 import { Kysely, Migrator } from 'kysely'
 
-import type { DatabaseDriverName } from '~/db-drivers'
-import { getDatabaseDriver } from '~/db-drivers'
+import type { DatabaseDriverName } from '~/lib/db/drivers'
+import { getDatabaseDriver } from '~/lib/db/drivers'
 import { migrations } from '~/lib/db/migrations'
 import { ENV } from '~/lib/env'
 import { logger } from '~/lib/logger'
 
 import type { Selectable } from 'kysely'
 
-export interface Database {
-  cache_keys: CacheKeysTable
-}
-
 export interface CacheKeysTable {
   key: string
   version: string
   updated_at: string
   accessed_at: string
 }
+export interface UploadsTable {
+  created_at: string
+  key: string
+  version: string
+  id: number
+  driver_upload_id: string
+}
+export interface UploadPartsTable {
+  upload_id: number
+  part_number: number
+  e_tag: string | null
+}
+
+export interface Database {
+  cache_keys: CacheKeysTable
+  uploads: UploadsTable
+  upload_parts: UploadPartsTable
+}
 
-let db: Kysely<Database>
+let _db: Kysely<Database>
 
 export async function initializeDatabase() {
   const driverName = ENV.DB_DRIVER
@@ -33,13 +47,13 @@ export async function initializeDatabase() {
 
   const driver = await driverSetup()
 
-  db = new Kysely<Database>({
+  _db = new Kysely<Database>({
     dialect: driver,
   })
 
   logger.info('Migrating database...')
   const migrator = new Migrator({
-    db,
+    db: _db,
     provider: {
       async getMigrations() {
         return migrations(driverName as DatabaseDriverName)
@@ -56,15 +70,24 @@ export async function initializeDatabase() {
   logger.success('Database migrated')
 }
 
+export function useDB() {
+  return _db
+}
+
+type DB = typeof _db
+
 /**
  * @see https://docs.github.com/en/actions/using-workflows/caching-dependencies-to-speed-up-workflows#matching-a-cache-key
  */
-export async function findKeyMatch(opts: { key: string; version: string; restoreKeys?: string[] }) {
-  logger.debug('Finding key match', opts)
+export async function findKeyMatch(
+  db: DB,
+  args: { key: string; version: string; restoreKeys?: string[] },
+) {
+  logger.debug('Finding key match', args)
   const exactPrimaryMatch = await db
     .selectFrom('cache_keys')
-    .where('key', '=', opts.key)
-    .where('version', '=', opts.version)
+    .where('key', '=', args.key)
+    .where('version', '=', args.version)
     .selectAll()
     .executeTakeFirst()
   if (exactPrimaryMatch) {
@@ -75,8 +98,8 @@ export async function findKeyMatch(opts: { key: string; version: string; restore
 
   const prefixedPrimaryMatch = await db
     .selectFrom('cache_keys')
-    .where('key', 'like', `${opts.key}%`)
-    .where('version', '=', opts.version)
+    .where('key', 'like', `${args.key}%`)
+    .where('version', '=', args.version)
     .orderBy('cache_keys.updated_at desc')
     .selectAll()
     .executeTakeFirst()
@@ -85,16 +108,16 @@ export async function findKeyMatch(opts: { key: string; version: string; restore
     return prefixedPrimaryMatch
   }
 
-  if (!opts.restoreKeys) {
+  if (!args.restoreKeys) {
     logger.debug('No restore keys provided')
     return
   }
 
-  logger.debug('Trying restore keys', opts.restoreKeys)
-  for (const key of opts.restoreKeys) {
+  logger.debug('Trying restore keys', args.restoreKeys)
+  for (const key of args.restoreKeys) {
     const exactMatch = await db
       .selectFrom('cache_keys')
-      .where('version', '=', opts.version)
+      .where('version', '=', args.version)
       .where('key', '=', key)
       .orderBy('cache_keys.updated_at desc')
       .selectAll()
@@ -107,7 +130,7 @@ export async function findKeyMatch(opts: { key: string; version: string; restore
 
     const prefixedMatch = await db
       .selectFrom('cache_keys')
-      .where('version', '=', opts.version)
+      .where('version', '=', args.version)
       .where('key', 'like', `${key}%`)
       .orderBy('cache_keys.updated_at desc')
       .selectAll()
@@ -121,7 +144,18 @@ export async function findKeyMatch(opts: { key: string; version: string; restore
   }
 }
 
-export async function updateOrCreateKey(key: string, version: string, date?: Date) {
+export async function updateOrCreateKey(
+  db: DB,
+  {
+    key,
+    version,
+    date,
+  }: {
+    key: string
+    version: string
+    date?: Date
+  },
+) {
   const now = date ?? new Date()
   const updateResult = await db
     .updateTable('cache_keys')
@@ -131,11 +165,14 @@ export async function updateOrCreateKey(key: string, version: string, date?: Dat
     .where('version', '=', version)
     .executeTakeFirst()
   if (Number(updateResult.numUpdatedRows) === 0) {
-    await createKey(key, version, date)
+    await createKey(db, { key, version, date })
   }
 }
 
-export async function touchKey(key: string, version: string, date?: Date) {
+export async function touchKey(
+  db: DB,
+  { key, version, date }: { key: string; version: string; date?: Date },
+) {
   const now = date ?? new Date()
   await db
     .updateTable('cache_keys')
@@ -145,7 +182,10 @@ export async function touchKey(key: string, version: string, date?: Date) {
     .execute()
 }
 
-export async function findStaleKeys(olderThanDays: number | undefined, date?: Date) {
+export async function findStaleKeys(
+  db: DB,
+  { olderThanDays, date }: { olderThanDays?: number; date?: Date },
+) {
   if (olderThanDays === undefined) return db.selectFrom('cache_keys').selectAll().execute()
 
   const now = date ?? new Date()
@@ -157,7 +197,10 @@ export async function findStaleKeys(olderThanDays: number | undefined, date?: Da
     .execute()
 }
 
-export async function createKey(key: string, version: string, date?: Date) {
+export async function createKey(
+  db: DB,
+  { key, version, date }: { key: string; version: string; date?: Date },
+) {
   const now = date ?? new Date()
   await db
     .insertInto('cache_keys')
@@ -170,10 +213,10 @@ export async function createKey(key: string, version: string, date?: Date) {
     .execute()
 }
 
-export async function pruneKeys(keys?: Selectable<CacheKeysTable>[]) {
+export async function pruneKeys(db: DB, keys?: Selectable<CacheKeysTable>[]) {
   if (keys) {
     await db.transaction().execute(async (tx) => {
-      for (const { key, version } of keys) {
+      for (const { key, version } of keys ?? []) {
         await tx
           .deleteFrom('cache_keys')
           .where('key', '=', key)
@@ -185,3 +228,13 @@ export async function pruneKeys(keys?: Selectable<CacheKeysTable>[]) {
     await db.deleteFrom('cache_keys').execute()
   }
 }
+
+export async function uploadExists(db: DB, { key, version }: { key: string; version: string }) {
+  const row = await db
+    .selectFrom('uploads')
+    .select('id')
+    .where('key', '=', key)
+    .where('version', '=', version)
+    .executeTakeFirst()
+  return !!row
+}

lib/db/migrations.ts

@@ -1,6 +1,6 @@
 import { type Migration, sql } from 'kysely'
 
-import type { DatabaseDriverName } from '~/db-drivers'
+import type { DatabaseDriverName } from '~/lib/db/drivers'
 
 export function migrations(dbType: DatabaseDriverName) {
   return {
@@ -24,5 +24,38 @@ export function migrations(dbType: DatabaseDriverName) {
         await db.schema.dropTable('cache_keys').ifExists().execute()
       },
     },
+    '2024-10-15T16:39:29': {
+      async up(db) {
+        await db.schema
+          .createTable('uploads')
+          .addColumn('created_at', 'text', (col) => col.notNull())
+          .addColumn('key', 'text', (col) => col.notNull())
+          .addColumn('version', 'text', (col) => col.notNull())
+          .addColumn('driver_upload_id', 'text', (col) => col.notNull())
+          .addColumn('id', 'integer', (col) => col.notNull())
+          .addPrimaryKeyConstraint('pk', ['id'])
+          .addUniqueConstraint('key_version', ['key', 'version'])
+          .ifNotExists()
+          .execute()
+        await db.schema
+          .createTable('upload_parts')
+          .addColumn('upload_id', 'integer')
+          .addColumn('part_number', 'integer')
+          .addColumn('e_tag', 'text')
+          .addPrimaryKeyConstraint('pk', ['upload_id', 'part_number'])
+          .addForeignKeyConstraint(
+            'fk_upload_parts_uploads',
+            ['upload_id'],
+            'uploads',
+            ['id'],
+            (c) => c.onDelete('cascade'),
+          )
+          .ifNotExists()
+          .execute()
+      },
+      async down(db) {
+        await db.schema.dropTable('uploads').ifExists().execute()
+      },
+    },
   } satisfies Record<string, Migration>
 }