webamp/packages/skin-database/api/processUserUploads.ts

import * as Skins from "../data/skins";
import S3 from "../s3";
import { addSkinFromBuffer } from "../addSkin";
import { EventHandler } from "./types";
import DiscordEventHandler from "./DiscordEventHandler";

async function* reportedUploads(): AsyncGenerator<
  { skin_md5: string; id: string; filename: string },
  void,
  unknown
> {
  const seen = new Set();
  while (true) {
    const upload = await Skins.getReportedUpload();
    console.log("Found one", { upload });
    if (upload == null) {
      return;
    }
    if (seen.has(upload.id)) {
      await Skins.recordUserUploadErrored(upload.id);
      console.error("Saw the same upload twice. It didn't get handled?");
      return;
    }
    seen.add(upload.id);
    yield upload;
  }
}

let processing = false;

function log(...args: any[]) {
  console.log(...args);
}

const ONE_MINUTE_IN_MS = 1000 * 60;

function timeout<T>(p: Promise<T>, duration: number): Promise<T> {
  return Promise.race([
    p,
    new Promise<never>((_resolve, reject) => {
      setTimeout(() => reject("timeout"), duration);
    }),
  ]);
}

async function processGivenUserUploads(
  eventHandler: EventHandler,
  uploads: AsyncGenerator<{ skin_md5: string; id: string; filename: string }>
) {
  log("Uploads to process...");
  for await (const upload of uploads) {
    log("Going to try: ", upload);
    try {
      const buffer = await S3.getUploadedSkin(upload.id);
      log("Got buffer: ", upload);
      const result = await timeout(
        addSkinFromBuffer(buffer, upload.filename, "Web API"),
        ONE_MINUTE_IN_MS
      );
      log("Added skin from buffer: ", upload, result);
      await Skins.recordUserUploadArchived(upload.id);
      log("Recorded upload archived: ", upload);
      if (result.status === "ADDED") {
        const action = {
          type:
            result.skinType === "CLASSIC"
              ? "CLASSIC_SKIN_UPLOADED"
              : "MODERN_SKIN_UPLOADED",
          md5: result.md5,
        } as const;
        log("Skin was added, sending action:", action);
        eventHandler(action);
        log("Action sent:", action);
      }
    } catch (e) {
      log("Upload errored, going to report it:", upload);
      await Skins.recordUserUploadErrored(upload.id);
      log("Recorded upload errored:", upload);
      const action = {
        type: "SKIN_UPLOAD_ERROR",
        uploadId: upload.id,
        message: e.message,
      } as const;
      eventHandler(action);
      console.error(e);
    }
  }
  log("Done processing uploads.");
}

export async function reprocessFailedUploads(handler: DiscordEventHandler) {
  // eslint-disable-next-line no-inner-declarations
  async function* erroredUploads(): AsyncGenerator<
    { skin_md5: string; id: string; filename: string },
    void,
    unknown
  > {
    const seen = new Set();
    while (true) {
      const upload = await Skins.getErroredUpload();
      console.log("Found one", { upload });
      if (upload == null) {
        return;
      }
      if (seen.has(upload.id)) {
        console.error("Saw the same upload twice. It didn't get handled?");
        return;
      }
      seen.add(upload.id);
      yield upload;
    }
  }
  const uploads = erroredUploads();

  await processGivenUserUploads((event) => handler.handle(event), uploads);
}

export async function processUserUploads(eventHandler: EventHandler) {
  log("process user uploads");
  // Ensure we only have one worker processing requests.
  if (processing) {
    return;
  }
  processing = true;
  const uploads = reportedUploads();
  await processGivenUserUploads(eventHandler, uploads);
  processing = false;
}