Refactor scraper paths; add websockets & showtime

Rename scraper modules to _generic/_post-processing and update imports across services. Add WebSocket support to app-ffmpeg (new lib/ws and /ws route) and hook into showtime V2. Improve bucket logging and upload flow; refactor find_seed_file into db/bucket/local runners, document seed files after upload, and trigger post-upload recording. Also change proxy lookup env from 'serverless' to 'edgeone' and update related scrape-worker imports.

src/index.ts

@@ -10,6 +10,7 @@ import { TestRouter } from "./sub-router/test";
 import { NS_runtime } from "../../../../packages/lib/env.runtime";
 import { NS_workers_proxy } from "../../../../packages/lib/proxy-server/workers-proxy";
 import { start_mongo_db } from "../../../../packages/db/mongo-db/init";
+import { NS_websockets } from "./lib/ws";
 
 await NS_runtime.set_env_var_db();
 
@@ -105,6 +106,31 @@ const app = new Elysia()
     return res;
   })
 
+  .ws("/ws", {
+    open(socket: any) {
+      const socket_id = socket.id;
+      const remote_address = socket.remoteAddress;
+
+      NS_websockets.ws_clients.set(socket_id, socket);
+
+      if (!NS_runtime.IS_PROD) {
+        console.log("WS CONNECTED:", remote_address, socket_id);
+      }
+
+      socket.send(JSON.stringify({ type: "CONNECTED", ws_id: socket_id }));
+    },
+    close(socket: any) {
+      const socket_id = socket.id;
+      const remote_address = socket.remoteAddress;
+
+      NS_websockets.ws_clients.delete(socket_id);
+
+      if (!NS_runtime.IS_PROD) {
+        console.log("WS CLOSED:", remote_address, socket_id);
+      }
+    }
+  })
+
   .listen({
     port: PORT || 7860,
     ...(NS_runtime.IS_PROD && { hostname: "0.0.0.0" }),

src/lib/bucket.ts

@@ -270,7 +270,7 @@ async function upload_to_bucket(arg0: {
     created_at: Date.now()
   });
 
-  console.log("\n", "🧲 UPLOADING TO BUCKET", "\n");
+  console.log("\n", `🧲 UPLOADING TO BUCKET: ${job_id}`, "\n");
 
   if (!skip?.probe) {
     const [E_, R_] = await NS_promise.F_catch_promise({
@@ -285,7 +285,7 @@ async function upload_to_bucket(arg0: {
 
     if (E_ || !R_) {
       upload_jobs.delete(job_id);
-      console.log("\n", "❌ UPLOAD BUCKET: FFPROBE FAILED", "\n");
+      console.log("\n", `❌ UPLOAD BUCKET: FFPROBE FAILED: ${job_id}`, "\n");
       return;
     }
   }
@@ -308,16 +308,11 @@ async function upload_to_bucket(arg0: {
   upload_jobs.delete(job_id);
 
   if (E__) {
-    console.log("\n", "❌ UPLOAD BUCKET FAILED", E__, "\n");
+    console.log("\n", `❌ UPLOAD BUCKET FAILED: ${job_id}`, E__, "\n");
     return { success: false };
   }
 
-  console.log("\n", "🔥 UPLOAD BUCKET DONE", "\n");
-
-  if (REQ0) {
-    REQ0.searchParams.set("ops", "just_check");
-    fetch(REQ0.toString());
-  }
+  console.log("\n", `🔥 UPLOAD BUCKET DONE: ${job_id}`, "\n");
 
   return { success: true };
 }

src/lib/ws.ts

@@ -0,0 +1,7 @@
+import { ServerWebSocket } from "elysia/ws/bun";
+
+const ws_clients = new Map<string, ServerWebSocket>();
+
+export const NS_websockets = {
+  ws_clients,
+};

src/sub-router/showtime/helper.ts

@@ -61,12 +61,14 @@ async function find_seed_file(arg0: {
     picture_quality = picture_quality.trim();
   }
 
-  // check database first
-  const db_ = await (async () => {
+  // check database
+  async function db_check_runner(arg0: { skip_db?: boolean }) {
+    const { skip_db } = arg0;
+
+    if (skip_db === true) return;
+
     const [_, db_file_info] = await NS_promise.F_catch_promise({
       promise: (async () => {
-        if (skip?.db === true) return;
-
         const _ =
           await NS_mongo_db_record_factory.NS_record_factory_s3_bucket.findRecordsByFingerprint(
             { job_id, sbe_id },
@@ -87,7 +89,7 @@ async function find_seed_file(arg0: {
 
     if (!in_db) return;
 
-    const stream_link = db_file_info!.url;
+    const stream_link = db_file_info?.url;
 
     const _u = new URL(stream_link);
     _u.searchParams.set("download", "true");
@@ -107,18 +109,22 @@ async function find_seed_file(arg0: {
       in_db,
       data
     };
-  })();
+  }
+
+  const db_ = await db_check_runner({ skip_db: skip?.db });
 
   if (db_?.data) {
     return db_.data;
   }
 
   // check bucket
-  const bucket_ = await (async () => {
+  async function bucket_check_runner(arg0: { skip_bucket?: boolean }) {
+    const { skip_bucket } = arg0;
+
+    if (skip_bucket === true) return;
+
     const [__, bucket_file_info] = await NS_promise.F_catch_promise({
       promise: (async () => {
-        if (skip?.bucket === true) return;
-
         return NS_bucket.get_latest_hf_bucket_file_by_name({
           folder_path: cache_folder,
           human_filename: file_id,
@@ -135,7 +141,20 @@ async function find_seed_file(arg0: {
 
     const stream_link = bucket_file_info!.url;
 
-    if (!db_?.in_db && skip?.db !== true && document === true) {
+    const document_metadata = await (async () => {
+      if (document !== true || db_?.in_db) return;
+
+      if (skip?.db === true) {
+        // const db_ = await db_check_runner({});
+        // if (db_?.in_db) {
+        //   return
+        // }
+      }
+
+      return true;
+    })();
+
+    if (document_metadata) {
       document_seed_file({
         job_id,
         sbe_id,
@@ -152,15 +171,19 @@ async function find_seed_file(arg0: {
       stream_link,
       size: bucket_file_info?.size
     };
-  })();
+  }
+
+  const bucket_ = await bucket_check_runner({ skip_bucket: skip?.bucket });
 
   if (bucket_) {
     return bucket_;
   }
 
   // check local
-  const local_ = await (async () => {
-    if (skip?.local === true) return;
+  async function locals_check_runner(arg0: { skip_local?: boolean }) {
+    const { skip_local } = arg0;
+
+    if (skip_local === true) return;
 
     const file =
       output_destination?.file_with_path &&
@@ -296,17 +319,15 @@ async function find_seed_file(arg0: {
     if (local_file) {
       return local_file;
     }
-  })();
+  }
+
+  const local_ = await locals_check_runner({ skip_local: skip?.local });
 
   if (local_) {
     return local_;
   }
 }
 
-async function find_seed_local_file(arg0: {}) {
-  const {} = arg0;
-}
-
 async function document_seed_file(arg0: {
   job_id: string;
   sbe_id: string;
@@ -328,10 +349,12 @@ async function document_seed_file(arg0: {
     qLevel
   } = arg0;
 
+  console.log("\n", `📊 DOCUMENTING: ${job_id}`, "\n");
+
   const [E_, R_] = await NS_promise.F_catch_promise({
     promise:
       NS_mongo_db_record_factory.NS_record_factory_s3_bucket.upsertRecord(
-        { job_id, sbe_id, file_id },
+        { job_id, sbe_id },
         {
           provider: "hugging face",
           job_id,

src/sub-router/showtime/v2.ts

@@ -15,6 +15,7 @@ import { spawn } from "node:child_process";
 import { NS_showtime_v2_helper } from "./helper";
 import { T_playback } from "../ffmpeg/validator-v2";
 import { NS_runtime } from "../../../../../../packages/lib/env.runtime";
+import { NS_websockets } from "../../lib/ws";
 
 const APP_ENV = () => NS_runtime.ENV_VAR_DB?.APP_ENV || process?.env?.APP_ENV;
 
@@ -35,6 +36,10 @@ type T_TranscodeJob = {
   error?: string;
   logs?: string[];
   log_errors?: string[];
+  db_metadata?: {
+    stream_link: string;
+    size: number | undefined;
+  };
   created_at: number;
 };
 
@@ -252,8 +257,16 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
       }
     }
 
-    // check/serve seeded file
-    const seed_file = await (async () => {
+    async function wide_search_runner(arg0: {
+      skip?: {
+        local?: boolean;
+        db?: boolean;
+        bucket?: boolean;
+      };
+      document: boolean;
+    }) {
+      const { skip, document } = arg0;
+
       if (!is_download_mode) return;
 
       const [E_, R_] = await NS_promise.F_catch_promise({
@@ -271,19 +284,12 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
           job_id,
           S3Dir,
           cleanup_corruption: true,
-          document: true,
-          ideal_runtime: playback.ideal_runtime
+          document,
+          ideal_runtime: playback.ideal_runtime,
+          skip
         })
       });
 
-      if (typeof R_ === "string") {
-        if (!NS_misc.isValidUrlFormat(R_)) {
-          return;
-        }
-
-        // create a function to proxima the seed link
-      }
-
       if (R_ && "stream_link" in R_ && R_.stream_link) {
         if (!NS_misc.isValidUrlFormat(R_.stream_link)) {
           return;
@@ -291,20 +297,48 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
       }
 
       return R_;
+    }
+
+    // check/serve seeded file
+    const seed_file = await (async () => {
+      const _ = current_job?.db_metadata?.stream_link;
+
+      if (_) {
+        return current_job.db_metadata;
+      }
+
+      const __ = await wide_search_runner({ document: true });
+
+      return __;
     })();
 
     if (seed_file) {
-      if (ops === "just_check") {
-        const file_size_bytes = parseFloat(`${seed_file?.size}`);
+      const file_size_bytes = parseFloat(`${seed_file?.size}`);
 
-        const percent = 100;
-        const percentage = typeof percent === "number" ? `${percent} %` : "";
-        const transcoded_size = file_size_bytes
-          ? NS_ffmpeg.F_format_size(file_size_bytes)
-          : current_job?.transcoded_size;
-        const progress_endpoint =
-          current_job?.progress_endpoint || progress_update_endpoint;
+      const percent = 100;
+      const percentage = typeof percent === "number" ? `${percent} %` : "";
+      const transcoded_size = file_size_bytes
+        ? NS_ffmpeg.F_format_size(file_size_bytes)
+        : current_job?.transcoded_size;
+      const progress_endpoint =
+        current_job?.progress_endpoint || progress_update_endpoint;
 
+      if (!current_job?.db_metadata?.stream_link) {
+        record_task_info({
+          job_id,
+          file_with_path: file_with_path || "",
+          status: "done",
+          info: {
+            percent,
+            transcoded_size,
+            ...("stream_link" in seed_file && seed_file.stream_link
+              ? { db_metadata: seed_file }
+              : {})
+          }
+        });
+      }
+
+      if (ops === "just_check") {
         return {
           passed: true,
           ready: true,
@@ -313,7 +347,6 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
           output_destination,
           flags: current_job?.flags,
           logs: current_job?.logs,
-          ...(typeof seed_file === "string" ? { stream_link: seed_file } : {}),
           ...(seed_file && "stream_link" in seed_file && seed_file.stream_link
             ? { stream_link: seed_file.stream_link }
             : {}),
@@ -322,12 +355,8 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
       }
 
       // handle direct file or url redirect
-      if (
-        seed_file &&
-        (typeof seed_file === "string" ||
-          ("stream_link" in seed_file && seed_file.stream_link))
-      ) {
-        const stream_link = new URL(seed_file.stream_link || seed_file);
+      if (seed_file && "stream_link" in seed_file && seed_file.stream_link) {
+        const stream_link = new URL(seed_file.stream_link);
         stream_link.searchParams.set("download", "true");
 
         const [E_, R_] = await NS_promise.F_catch_promise({
@@ -473,10 +502,6 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
       }
     }
 
-    let node_proc = spawn(NS_ffmpeg.ffmpeg_path, flags, {
-      signal: controller.signal
-    });
-
     const passthrough = (() => {
       if (delivery_mode !== "stream") {
         return;
@@ -486,6 +511,10 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
       return new PassThrough();
     })();
 
+    let node_proc = spawn(NS_ffmpeg.ffmpeg_path, flags, {
+      signal: controller.signal
+    });
+
     let progress_data: Record<string, any> | undefined = undefined;
     let seek_size_bytes = 0;
 
@@ -648,11 +677,30 @@ const sub_router = new Elysia({ prefix: "/showtimeV2" }).get(
               file_with_path &&
               APP_ENV() !== "dev"
             ) {
+              transcode_jobs.delete(job_id);
+
               NS_bucket.upload_to_bucket({
                 job_id,
                 bucket_path: `${cache_folder}/${file_id}`,
                 file_with_path,
                 REQ0
+              }).then((cc) => {
+                if (!cc?.success) {
+                  return;
+                }
+
+                // document file
+                wide_search_runner({
+                  document: true,
+                  skip: {
+                    local: true,
+                    db: true
+                  }
+                }).then((ccc) => {
+                  if (!ccc) {
+                    return;
+                  }
+                });
               });
             }
           }