feat: storage cleanup job (#29)

Author: LouisHaftmann

.gitignore

@@ -10,4 +10,4 @@ data/
 .data
 .idea
 .DS_Store
-.test-data/
+.test-data/

docs/content/1.getting-started/1.index.md

@@ -49,6 +49,12 @@ The storage driver to use for storing cache data. For more information, see [Sto
 
 The database driver to use for storing cache metadata. For more information, see [Database Drivers](/getting-started/database-drivers).
 
+#### `CLEANUP_OLDER_THAN_DAYS`
+
+- Default: `90`
+
+The number of days to keep stale cache data before deleting it. Set to `0` to disable cache cleanup.
+
 #### `NITRO_PORT`
 
 - Default: `3000`

lib/db/index.ts

@@ -6,6 +6,8 @@ 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
 }
@@ -14,6 +16,7 @@ export interface CacheKeysTable {
   key: string
   version: string
   updated_at: string
+  accessed_at: string
 }
 
 async function initializeDatabase() {
@@ -106,31 +109,65 @@ export async function findKeyMatch(opts: { key: string; version: string; restore
   }
 }
 
-export async function touchKey(key: string, version: string) {
-  const now = new Date()
+export async function updateOrCreateKey(key: string, version: string, date?: Date) {
+  const now = date ?? new Date()
   const updateResult = await db
     .updateTable('cache_keys')
     .set('updated_at', now.toISOString())
+    .set('accessed_at', now.toISOString())
     .where('key', '=', key)
     .where('version', '=', version)
     .executeTakeFirst()
   if (Number(updateResult.numUpdatedRows) === 0) {
-    await createKey(key, version)
+    await createKey(key, version, date)
   }
 }
 
-export async function createKey(key: string, version: string) {
-  const now = new Date()
+export async function touchKey(key: string, version: string, date?: Date) {
+  const now = date ?? new Date()
+  await db
+    .updateTable('cache_keys')
+    .set('accessed_at', now.toISOString())
+    .where('key', '=', key)
+    .where('version', '=', version)
+    .execute()
+}
+
+export async function findStaleKeys(olderThanDays: number | undefined, date?: Date) {
+  if (olderThanDays === undefined) return db.selectFrom('cache_keys').selectAll().execute()
+
+  const now = date ?? new Date()
+  const threshold = new Date(now.getTime() - olderThanDays * 24 * 60 * 60 * 1000)
+  return db
+    .selectFrom('cache_keys')
+    .where('accessed_at', '<', threshold.toISOString())
+    .selectAll()
+    .execute()
+}
+
+export async function createKey(key: string, version: string, date?: Date) {
+  const now = date ?? new Date()
   await db
     .insertInto('cache_keys')
     .values({
       key,
       version,
       updated_at: now.toISOString(),
+      accessed_at: now.toISOString(),
     })
     .execute()
 }
 
-export async function pruneKeys() {
-  await db.deleteFrom('cache_keys').execute()
+export async function pruneKeys(keys?: Selectable<CacheKeysTable>[]) {
+  if (!keys) await db.deleteFrom('cache_keys').execute()
+  else
+    await db.transaction().execute(async (tx) => {
+      for (const { key, version } of keys) {
+        await tx
+          .deleteFrom('cache_keys')
+          .where('key', '=', key)
+          .where('version', '=', version)
+          .execute()
+      }
+    })
 }

lib/db/migrations.ts

@@ -13,6 +13,7 @@ export const migrations = (dbType: DatabaseDriverName) =>
             col.notNull(),
           )
           .addColumn('updated_at', 'text', (col) => col.notNull())
+          .addColumn('accessed_at', 'text', (col) => col.notNull())
           .addPrimaryKeyConstraint('pk', ['key', 'version'])
 
         if (dbType === 'mysql') query = query.modifyEnd(sql`engine=InnoDB CHARSET=latin1`)

lib/env.ts

@@ -4,6 +4,7 @@ import { logger } from '~/lib/logger'
 
 const envSchema = z.object({
   URL_ACCESS_TOKEN: z.string().min(1),
+  CLEANUP_OLDER_THAN_DAYS: z.coerce.number().int().min(1).default(90),
   API_BASE_URL: z.string().min(1),
   STORAGE_DRIVER: z.string().toLowerCase().default('filesystem'),
   DB_DRIVER: z.string().toLowerCase().default('sqlite'),

lib/storage/index.ts

@@ -3,7 +3,7 @@ import { createHash, randomBytes, randomInt } from 'node:crypto'
 
 import consola from 'consola'
 
-import { findKeyMatch, touchKey } from '~/lib/db'
+import { findKeyMatch, findStaleKeys, pruneKeys, touchKey, updateOrCreateKey } from '~/lib/db'
 import { ENV } from '~/lib/env'
 import { logger } from '~/lib/logger'
 import { encodeCacheKey } from '~/lib/storage/driver'
@@ -30,7 +30,7 @@ async function initializeStorageDriver() {
 
     return <StorageAdapter>{
       async reserveCache(key, version, cacheSize) {
-        logger.debug('Reserve: Trying to reserve cache for', key, version, cacheSize)
+        logger.debug('Reserve: Reserving cache for', key, version, cacheSize)
         const bufferKey = `${key}:${version}`
         const existingBuffer = uploadBuffers.get(bufferKey)
         if (existingBuffer) {
@@ -59,7 +59,7 @@ async function initializeStorageDriver() {
         }
       },
       async getCacheEntry(keys, version) {
-        logger.debug('Get: Trying to get cache entry for', keys, version)
+        logger.debug('Get: Getting cache entry for', keys, version)
         const primaryKey = keys[0]
         const restoreKeys = keys.length > 1 ? keys.slice(1) : undefined
 
@@ -70,6 +70,8 @@ async function initializeStorageDriver() {
           return null
         }
 
+        await touchKey(cacheKey.key, cacheKey.version)
+
         const cacheFileName = encodeCacheKey(cacheKey.key, cacheKey.version)
         const hashedKey = createHash('sha256')
           .update(cacheFileName + DOWNLOAD_SECRET_KEY)
@@ -83,7 +85,7 @@ async function initializeStorageDriver() {
         }
       },
       async commitCache(uploadId) {
-        logger.debug('Commit: Trying to commit cache for upload', uploadId)
+        logger.debug('Commit: Committing cache for upload', uploadId)
 
         if (commitLocks.has(uploadId)) {
           logger.debug(`Commit: Commit for upload ${uploadId} already in progress. Ignoring...`)
@@ -109,7 +111,7 @@ async function initializeStorageDriver() {
         try {
           logger.debug('Commit: Committing cache for id', uploadId)
           await driver.upload(buffer, cacheFileName)
-          await touchKey(cacheKey.key, cacheKey.version)
+          await updateOrCreateKey(cacheKey.key, cacheKey.version)
           logger.debug('Commit: Cache committed for id', uploadId)
         } finally {
           cacheKeyByUploadId.delete(uploadId)
@@ -118,11 +120,11 @@ async function initializeStorageDriver() {
         }
       },
       async download(objectName) {
-        logger.debug('Download: Trying to download', objectName)
+        logger.debug('Download: Downloading', objectName)
         return driver.download(objectName)
       },
       async uploadChunk(uploadId, chunkStream, chunkStart) {
-        logger.debug('Upload: Trying to upload chunk for upload', uploadId)
+        logger.debug('Upload: Uploading chunk for upload', uploadId)
         const cacheKey = cacheKeyByUploadId.get(uploadId)
         if (!cacheKey) {
           logger.debug(`Upload: No cache key found for upload ${uploadId}. Ignoring...`)
@@ -148,9 +150,13 @@ async function initializeStorageDriver() {
         await chunkStream.pipeTo(bufferWriteStream)
         logger.debug('Upload: Chunks uploaded for id', uploadId)
       },
-      async pruneCaches() {
-        logger.debug('Prune: Trying to prune caches')
-        await driver.prune()
+      async pruneCaches(olderThanDays) {
+        logger.debug('Prune: Pruning caches')
+
+        const keys = await findStaleKeys(olderThanDays)
+        await driver.delete(keys.map((key) => encodeCacheKey(key.key, key.version)))
+        await pruneKeys(keys)
+
         logger.debug('Prune: Caches pruned')
       },
     }

lib/types.ts

@@ -4,7 +4,7 @@ import type { Readable } from 'node:stream'
 export interface StorageDriver {
   upload: (buffer: Buffer, objectName: string) => Promise<void> | void
   download: (objectName: string) => Promise<Readable> | Readable
-  prune: () => Promise<void> | void
+  delete: (objectNames: string[]) => Promise<void> | void
 }
 
 export interface StorageAdapter {
@@ -18,7 +18,7 @@ export interface StorageAdapter {
   ) => Promise<void>
   commitCache: (uploadId: number, size: number) => Promise<void>
   reserveCache: (key: string, version: string, cacheSize: number) => Promise<ReserveCacheResponse>
-  pruneCaches: () => Promise<void>
+  pruneCaches: (olderThanDays?: number) => Promise<void>
 }
 
 export interface ArtifactCacheEntry {

package.json

@@ -3,6 +3,7 @@
   "type": "module",
   "version": "1.0.0",
   "private": true,
+  "packageManager": "pnpm@8.15.7",
   "scripts": {
     "prepare": "nitropack prepare && husky",
     "dev": "nitropack dev",
@@ -24,6 +25,7 @@
     "@types/pg": "^8.11.5",
     "better-sqlite3": "^9.5.0",
     "consola": "^3.2.3",
+    "croner": "^8.0.2",
     "kysely": "^0.27.3",
     "minio": "^7.1.3",
     "mysql2": "^3.9.7",

plugins/cleanup.ts

@@ -0,0 +1,18 @@
+import { colorize } from 'consola/utils'
+import { Cron } from 'croner'
+
+import { ENV } from '@/lib/env'
+import { logger } from '@/lib/logger'
+import { storageAdapter } from '@/lib/storage'
+
+export default defineNitroPlugin(() => {
+  // daily
+  const job = Cron('0 0 * * *')
+  const nextRun = job.nextRun()
+  logger.info(
+    `Cleaning up cache entries older than ${colorize('blue', `${ENV.CLEANUP_OLDER_THAN_DAYS}d`)} with schedule ${colorize('blue', job.getPattern() ?? '')}${nextRun ? ` (next run: ${nextRun.toLocaleString()})` : ''}`,
+  )
+  job.schedule(async () => {
+    await storageAdapter.pruneCaches(ENV.CLEANUP_OLDER_THAN_DAYS)
+  })
+})