packages/backend/root.ts

@@ -3,10 +3,10 @@ import { VERSION } from "./main.ts";
 import { createHandlers } from "./utils.ts";
 
 export const rootHandler = createHandlers((c) => {
-	let singer: Singer | Singer[];
+	let singer: Singer | Singer[] | null;
 	const shouldShowSpecialSinger = Math.random() < 0.016;
 	if (getSingerForBirthday().length !== 0) {
-		singer = JSON.parse(JSON.stringify(getSingerForBirthday())) as Singer[];
+		singer = getSingerForBirthday();
 		for (const s of singer) {
 			delete s.birthday;
 			s.message = `祝${s.name}生日快乐~`;

packages/crawler/db/snapshot.ts

@@ -6,16 +6,8 @@ export async function getVideosNearMilestone(client: Client) {
 	const queryResult = await client.queryObject<LatestSnapshotType>(`
         SELECT ls.*
         FROM latest_video_snapshot ls
-                 RIGHT JOIN songs ON songs.aid = ls.aid
         WHERE
-            (views >= 50000 AND views < 100000) OR
+            views < 100000 OR
-            (views >= 900000 AND views < 1000000) OR
-            (views >= 9900000 AND views < 10000000)
-        UNION
-        SELECT ls.*
-        FROM latest_video_snapshot ls
-        WHERE
-            (views >= 90000 AND views < 100000) OR
             (views >= 900000 AND views < 1000000) OR
             (views >= 9900000 AND views < 10000000)
     `);

packages/crawler/db/snapshotSchedule.ts

@@ -1,7 +1,7 @@
 import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 import { SnapshotScheduleType } from "@core/db/schema";
 import logger from "log/logger.ts";
-import { MINUTE } from "$std/datetime/constants.ts";
+import { MINUTE } from "@std/datetime";
 import { redis } from "@core/db/redis.ts";
 import { Redis } from "ioredis";
 
@@ -274,14 +274,8 @@ export async function getBulkSnapshotsInNextSecond(client: Client) {
 	const query = `
 		SELECT *
 		FROM snapshot_schedule
-        WHERE (started_at <= NOW() + INTERVAL '15 seconds')
+		WHERE started_at <= NOW() + INTERVAL '15 seconds' AND status = 'pending' AND type = 'normal'
-          AND status = 'pending'
+		ORDER BY started_at
-          AND (type = 'normal' OR type = 'archive')
-        ORDER BY CASE
-                     WHEN type = 'normal' THEN 1
-                     WHEN type = 'archive' THEN 2
-                     END,
-                 started_at
 		LIMIT 1000;
 	`;
 	const res = await client.queryObject<SnapshotScheduleType>(query, []);
@@ -312,14 +306,3 @@ export async function getVideosWithoutActiveSnapshotSchedule(client: Client) {
 	const res = await client.queryObject<{ aid: number }>(query, []);
 	return res.rows.map((r) => Number(r.aid));
 }
-
-export async function getAllVideosWithoutActiveSnapshotSchedule(client: Client) {
-	const query: string = `
-		SELECT s.aid
-		FROM bilibili_metadata s
-		LEFT JOIN snapshot_schedule ss ON s.aid = ss.aid AND (ss.status = 'pending' OR ss.status = 'processing')
-		WHERE ss.aid IS NULL
-	`;
-	const res = await client.queryObject<{ aid: number }>(query, []);
-	return res.rows.map((r) => Number(r.aid));
-}

packages/crawler/deno.json

@@ -12,7 +12,7 @@
 		"worker:filter": "deno run --env-file=.env --allow-env --allow-read --allow-ffi --allow-net --allow-write ./src/filterWorker.ts",
 		"adder": "deno run --env-file=.env --allow-env --allow-read --allow-ffi --allow-net ./src/jobAdder.ts",
 		"bullui": "deno run --allow-read --allow-env --allow-ffi --allow-net ./src/bullui.ts",
-		"all": "concurrently --restart-tries -1 'deno task worker:main' 'deno task adder' 'deno task bullui' 'deno task worker:filter'",
+		"all": "concurrently 'deno task worker:main' 'deno task adder' 'deno task bullui' 'deno task worker:filter'",
 		"test": "deno test ./test/ --allow-env --allow-ffi --allow-read --allow-net --allow-write --allow-run"
 	},
 	"lint": {

packages/crawler/mq/exec/archiveSnapshots.ts

@@ -1,40 +0,0 @@
-import { Job } from "npm:bullmq@5.45.2";
-import { getAllVideosWithoutActiveSnapshotSchedule, scheduleSnapshot } from "db/snapshotSchedule.ts";
-import { withDbConnection } from "db/withConnection.ts";
-import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
-import logger from "log/logger.ts";
-import { lockManager } from "mq/lockManager.ts";
-import { getLatestVideoSnapshot } from "db/snapshot.ts";
-import { HOUR, MINUTE } from "$std/datetime/constants.ts";
-
-export const archiveSnapshotsWorker = async (_job: Job) =>
-	await withDbConnection<void>(async (client: Client) => {
-		const startedAt = Date.now();
-		if (await lockManager.isLocked("dispatchArchiveSnapshots")) {
-			logger.log("dispatchArchiveSnapshots is already running", "mq");
-			return;
-		}
-		await lockManager.acquireLock("dispatchArchiveSnapshots", 30 * 60);
-		const aids = await getAllVideosWithoutActiveSnapshotSchedule(client);
-		for (const rawAid of aids) {
-			const aid = Number(rawAid);
-			const latestSnapshot = await getLatestVideoSnapshot(client, aid);
-			const now = Date.now();
-			const lastSnapshotedAt = latestSnapshot?.time ?? now;
-			const interval = 168;
-			logger.log(
-				`Scheduled archive snapshot for aid ${aid} in ${interval} hours.`,
-				"mq",
-				"fn:archiveSnapshotsWorker",
-			);
-			const targetTime = lastSnapshotedAt + interval * HOUR;
-			await scheduleSnapshot(client, aid, "archive", targetTime);
-			if (now - startedAt > 250 * MINUTE) {
-				return;
-			}
-		}
-	}, (e) => {
-		logger.error(e as Error, "mq", "fn:archiveSnapshotsWorker");
-	}, async () => {
-		await lockManager.releaseLock("dispatchArchiveSnapshots");
-	});

packages/crawler/mq/exec/executors.ts

@@ -2,4 +2,3 @@ export * from "mq/exec/getLatestVideos.ts";
 export * from "./getVideoInfo.ts";
 export * from "./collectSongs.ts";
 export * from "./takeBulkSnapshot.ts";
-export * from "./archiveSnapshots.ts";

packages/crawler/mq/exec/getLatestVideos.ts

@@ -3,7 +3,8 @@ import { queueLatestVideos } from "mq/task/queueLatestVideo.ts";
 import { withDbConnection } from "db/withConnection.ts";
 import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 
+
 export const getLatestVideosWorker = (_job: Job): Promise<void> =>
 	withDbConnection(async (client: Client) => {
-		await queueLatestVideos(client);
+		await queueLatestVideos(client)
 	});

packages/crawler/mq/exec/getVideoInfo.ts

@@ -11,5 +11,5 @@ export const getVideoInfoWorker = async (job: Job): Promise<void> =>
 			logger.warn("aid does not exists", "mq", "job:getVideoInfo");
 			return;
 		}
-		await insertVideoInfo(client, aid);
+		await insertVideoInfo(client, aid)
 	});

packages/crawler/mq/exec/snapshotTick.ts

@@ -47,19 +47,13 @@ export const bulkSnapshotTickWorker = async (_job: Job) => {
 			const filteredAids = await bulkGetVideosWithoutProcessingSchedules(client, aids);
 			if (filteredAids.length === 0) continue;
 			await bulkSetSnapshotStatus(client, filteredAids, "processing");
-			const schedulesData = group.map((schedule) => {
+			const dataMap: { [key: number]: number } = {};
-				return {
+			for (const schedule of group) {
-					aid: Number(schedule.aid),
+				const id = Number(schedule.id);
-					id: Number(schedule.id),
+				dataMap[id] = Number(schedule.aid);
-					type: schedule.type,
+			}
-					created_at: schedule.created_at,
-					started_at: schedule.started_at,
-					finished_at: schedule.finished_at,
-					status: schedule.status,
-				};
-			});
 			await SnapshotQueue.add("bulkSnapshotVideo", {
-				schedules: schedulesData,
+				map: dataMap,
 			}, { priority: 3 });
 		}
 		return `OK`;
@@ -85,7 +79,7 @@ export const snapshotTickWorker = async (_job: Job) => {
 			const aid = Number(schedule.aid);
 			await setSnapshotStatus(client, schedule.id, "processing");
 			await SnapshotQueue.add("snapshotVideo", {
-				aid: Number(aid),
+				aid: aid,
 				id: Number(schedule.id),
 				type: schedule.type ?? "normal",
 			}, { priority });
@@ -228,7 +222,7 @@ export const takeSnapshotForVideoWorker = async (job: Job) => {
 				"mq",
 				"fn:takeSnapshotForVideoWorker",
 			);
-			await setSnapshotStatus(client, id, "no_proxy");
+			await setSnapshotStatus(client, id, "completed");
 			await scheduleSnapshot(client, aid, type, Date.now() + retryInterval);
 			return;
 		}

packages/crawler/mq/exec/takeBulkSnapshot.ts

@@ -11,17 +11,15 @@ import logger from "log/logger.ts";
 import { NetSchedulerError } from "@core/net/delegate.ts";
 import { HOUR, MINUTE, SECOND } from "@std/datetime";
 import { getRegularSnapshotInterval } from "../task/regularSnapshotInterval.ts";
-import { SnapshotScheduleType } from "@core/db/schema";
 
 export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
-	const schedules: SnapshotScheduleType[] = job.data.schedules;
+	const dataMap: { [key: number]: number } = job.data.map;
-	const ids = schedules.map((schedule) => Number(schedule.id));
+	const ids = Object.keys(dataMap).map((id) => Number(id));
 	const aidsToFetch: number[] = [];
 	const client = await db.connect();
 	try {
-		for (const schedule of schedules) {
+		for (const id of ids) {
-			const aid = Number(schedule.aid);
+			const aid = Number(dataMap[id]);
-			const id = Number(schedule.id);
 			const exists = await snapshotScheduleExists(client, id);
 			if (!exists) {
 				continue;
@@ -56,11 +54,7 @@ export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
 			logger.log(`Taken snapshot for video ${aid} in bulk.`, "net", "fn:takeBulkSnapshotForVideosWorker");
 		}
 		await bulkSetSnapshotStatus(client, ids, "completed");
-
+		for (const aid of aidsToFetch) {
-		for (const schedule of schedules) {
-			const aid = Number(schedule.aid);
-			const type = schedule.type;
-			if (type == "archive") continue;
 			const interval = await getRegularSnapshotInterval(client, aid);
 			logger.log(`Scheduled regular snapshot for aid ${aid} in ${interval} hours.`, "mq");
 			await scheduleSnapshot(client, aid, "normal", Date.now() + interval * HOUR);
@@ -73,8 +67,8 @@ export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
 				"mq",
 				"fn:takeBulkSnapshotForVideosWorker",
 			);
-			await bulkSetSnapshotStatus(client, ids, "no_proxy");
+			await bulkSetSnapshotStatus(client, ids, "completed");
-			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 20 * MINUTE * Math.random());
+			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 2 * MINUTE);
 			return;
 		}
 		logger.error(e as Error, "mq", "fn:takeBulkSnapshotForVideosWorker");

packages/crawler/mq/init.ts

@@ -1,4 +1,4 @@
-import { HOUR, MINUTE, SECOND } from "$std/datetime/constants.ts";
+import { MINUTE, SECOND } from "@std/datetime";
 import { ClassifyVideoQueue, LatestVideosQueue, SnapshotQueue } from "mq/index.ts";
 import logger from "log/logger.ts";
 import { initSnapshotWindowCounts } from "db/snapshotSchedule.ts";
@@ -55,11 +55,6 @@ export async function initMQ() {
 			immediately: true,
 		});
 
-		await SnapshotQueue.upsertJobScheduler("dispatchArchiveSnapshots", {
-			every: 6 * HOUR,
-			immediately: true,
-		});
-
 		await SnapshotQueue.upsertJobScheduler("scheduleCleanup", {
 			every: 30 * MINUTE,
 			immediately: true,

packages/crawler/src/worker.ts

@@ -1,11 +1,5 @@
 import { ConnectionOptions, Job, Worker } from "bullmq";
-import {
+import { collectSongsWorker, getLatestVideosWorker, getVideoInfoWorker } from "mq/exec/executors.ts";
-	archiveSnapshotsWorker,
-	collectSongsWorker,
-	getLatestVideosWorker,
-	getVideoInfoWorker,
-	takeBulkSnapshotForVideosWorker,
-} from "mq/exec/executors.ts";
 import { redis } from "@core/db/redis.ts";
 import logger from "log/logger.ts";
 import { lockManager } from "mq/lockManager.ts";
@@ -18,22 +12,10 @@ import {
 	snapshotTickWorker,
 	takeSnapshotForVideoWorker,
 } from "mq/exec/snapshotTick.ts";
-
+import {takeBulkSnapshotForVideosWorker} from "mq/exec/executors.ts";
-const releaseLockForJob = async (name: string) => {
-	await lockManager.releaseLock(name);
-	logger.log(`Released lock: ${name}`, "mq");
-};
-
-const releaseAllLocks = async () => {
-	const locks = ["dispatchRegularSnapshots", "dispatchArchiveSnapshots", "getLatestVideos"];
-	for (const lock of locks) {
-		await releaseLockForJob(lock);
-	}
-};
 
 Deno.addSignalListener("SIGINT", async () => {
 	logger.log("SIGINT Received: Shutting down workers...", "mq");
-	await releaseAllLocks();
 	await latestVideoWorker.close(true);
 	await snapshotWorker.close(true);
 	Deno.exit();
@@ -41,7 +23,6 @@ Deno.addSignalListener("SIGINT", async () => {
 
 Deno.addSignalListener("SIGTERM", async () => {
 	logger.log("SIGTERM Received: Shutting down workers...", "mq");
-	await releaseAllLocks();
 	await latestVideoWorker.close(true);
 	await snapshotWorker.close(true);
 	Deno.exit();
@@ -78,6 +59,10 @@ latestVideoWorker.on("error", (err) => {
 	logger.error(e.rawError, e.service, e.codePath);
 });
 
+latestVideoWorker.on("closed", async () => {
+	await lockManager.releaseLock("getLatestVideos");
+});
+
 const snapshotWorker = new Worker(
 	"snapshot",
 	async (job: Job) => {
@@ -96,8 +81,6 @@ const snapshotWorker = new Worker(
 				return await takeBulkSnapshotForVideosWorker(job);
 			case "bulkSnapshotTick":
 				return await bulkSnapshotTickWorker(job);
-			case "dispatchArchiveSnapshots":
-				return await archiveSnapshotsWorker(job);
 			default:
 				break;
 		}
@@ -109,3 +92,7 @@ snapshotWorker.on("error", (err) => {
 	const e = err as WorkerError;
 	logger.error(e.rawError, e.service, e.codePath);
 });
+
+snapshotWorker.on("closed", async () => {
+	await lockManager.releaseLock("dispatchRegularSnapshots");
+});