Move file caching logic to new FileCache service

Also: removes the -re ffmpeg option.
If this option is passed, ffmpeg won't write to fs-capacitor (and the cache file) as fast as possible.
In other words, the cache file won't finish writing until the entire stream has been played.

1 files changed, 109 insertions, 0 deletions

diff --git a/src/services/file-cache.ts b/src/services/file-cache.ts
new file mode 100644
index 0000000..0ecf279
--- /dev/null
+++ b/src/services/file-cache.ts
@@ -0,0 +1,109 @@
+import {inject, injectable} from 'inversify';
+import sequelize from 'sequelize';
+import {FileCache} from '../models/index.js';
+import {TYPES} from '../types.js';
+import Config from './config.js';
+import {promises as fs, createWriteStream} from 'fs';
+import path from 'path';
+
+@injectable()
+export default class FileCacheProvider {
+  private readonly config: Config;
+
+  constructor(@inject(TYPES.Config) config: Config) {
+    this.config = config;
+  }
+
+  /**
+   * Returns path to cached file if it exists, otherwise throws an error.
+   * Updates the `accessedAt` property of the cached file.
+   * @param hash lookup key
+   */
+  async getPathFor(hash: string): Promise<string> {
+    const model = await FileCache.findByPk(hash);
+
+    if (!model) {
+      throw new Error('File is not cached');
+    }
+
+    const resolvedPath = path.join(this.config.CACHE_DIR, hash);
+
+    try {
+      await fs.access(resolvedPath);
+    } catch (_: unknown) {
+      await FileCache.destroy({where: {hash}});
+
+      throw new Error('File is not cached');
+    }
+
+    await model.update({accessedAt: new Date()});
+
+    return resolvedPath;
+  }
+
+  /**
+   * Returns a write stream for the given hash key.
+   * The stream handles saving a new file and will
+   * update the database after the stream is finished.
+   * @param hash lookup key
+   */
+  createWriteStream(hash: string) {
+    const tmpPath = path.join(this.config.CACHE_DIR, 'tmp', hash);
+    const finalPath = path.join(this.config.CACHE_DIR, hash);
+
+    const stream = createWriteStream(tmpPath);
+
+    stream.on('finish', async () => {
+      // Only move if size is non-zero (may have errored out)
+      const stats = await fs.stat(tmpPath);
+
+      if (stats.size !== 0) {
+        await fs.rename(tmpPath, finalPath);
+      }
+
+      await FileCache.create({hash, bytes: stats.size, accessedAt: new Date()});
+
+      await this.evictOldestIfNecessary();
+    });
+
+    return stream;
+  }
+
+  /**
+   * Deletes orphaned cache files and evicts files if
+   * necessary. Should be run on program startup so files
+   * will be evicted if the cache limit has changed.
+   */
+  async cleanup() {
+    await this.removeOrphans();
+    await this.evictOldestIfNecessary();
+  }
+
+  private async evictOldestIfNecessary() {
+    const [{dataValues: {totalSizeBytes}}] = await FileCache.findAll({
+      attributes: [
+        [sequelize.fn('sum', sequelize.col('bytes')), 'totalSizeBytes'],
+      ],
+    }) as unknown as [{dataValues: {totalSizeBytes: number}}];
+
+    if (totalSizeBytes > this.config.CACHE_LIMIT_IN_BYTES) {
+      const oldest = await FileCache.findOne({
+        order: [
+          ['accessedAt', 'ASC'],
+        ],
+      });
+
+      if (oldest) {
+        await oldest.destroy();
+        await fs.unlink(path.join(this.config.CACHE_DIR, oldest.hash));
+      }
+
+      // Continue to evict until we're under the limit
+      await this.evictOldestIfNecessary();
+    }
+  }
+
+  private async removeOrphans() {
+    // TODO
+  }
+}