.gitignore

@@ -39,6 +39,4 @@ redis/
 dist/
 build/
 
-docker-compose.yml
+docker-compose.yml
-
-ucaptcha-config.yaml

.idea/compiler.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="TypeScriptCompiler">
+    <option name="useTypesFromServer" value="true" />
+  </component>
+</project>

.idea/cvsa.iml

@@ -28,8 +28,6 @@
       <excludeFolder url="file://$MODULE_DIR$/packages/crawler/logs" />
       <excludeFolder url="file://$MODULE_DIR$/data" />
       <excludeFolder url="file://$MODULE_DIR$/redis" />
-      <excludeFolder url="file://$MODULE_DIR$/ml" />
-      <excludeFolder url="file://$MODULE_DIR$/src" />
     </content>
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />

.tokeignore

@@ -5,7 +5,4 @@ data
 *.md
 *config*
 Inter.css
-MiSans.css
+MiSans.css
-*.yaml
-*.yml
-*.mdx

Dockerfile.frontend

@@ -1,7 +1,5 @@
 FROM oven/bun
 
-ARG BACKEND_URL
-
 WORKDIR /app
 
 COPY . .
@@ -14,7 +12,6 @@ RUN bun run build
 
 ENV HOST=0.0.0.0
 ENV PORT=4321
-ENV BACKEND_URL=${BACKEND_URL}
 
 EXPOSE 4321
 

Dockerfile.next

@@ -1,14 +0,0 @@
-FROM node:lts-slim AS production
-
-WORKDIR /app
-
-COPY ./packages/next/.next ./.next
-COPY ./packages/next/public ./public
-COPY ./packages/next/package.json ./package.json
-COPY ./packages/next/node_modules ./node_modules
-
-ENV NODE_ENV production
-
-EXPOSE 7400
-
-CMD ["npm", "start"]

deno.json

@@ -0,0 +1,16 @@
+{
+	"lock": false,
+	"workspace": ["./packages/crawler", "./packages/core"],
+	"nodeModulesDir": "auto",
+	"tasks": {
+		"crawler": "deno task --filter 'crawler' all",
+		"backend": "deno task --filter 'backend' start"
+	},
+	"fmt": {
+		"useTabs": true,
+		"lineWidth": 120,
+		"indentWidth": 4,
+		"semiColons": true,
+		"proseWrap": "always"
+	}
+}

package.json

@@ -1,23 +1,20 @@
 {
-	"name": "cvsa",
+    "name": "cvsa",
-	"version": "3.15.34",
+    "version": "2.13.22",
-	"private": false,
+    "private": false,
-	"type": "module",
+    "type": "module",
-	"workspaces": [
+    "workspaces": [
-		"packages/frontend",
+        "packages/frontend",
-		"packages/core",
+        "packages/core",
-		"packages/backend",
+        "packages/backend",
-		"packages/crawler"
+        "packages/crawler"
-	],
+    ],
-	"dependencies": {
+    "dependencies": {
-		"arg": "^5.0.2",
+        "postgres": "^3.4.5"
-		"postgres": "^3.4.5"
+    },
-	},
+    "devDependencies": {
-	"devDependencies": {
+        "vite-tsconfig-paths": "^5.1.4",
-		"@types/bun": "^1.2.15",
+        "vitest": "^3.1.2",
-		"prettier": "^3.5.3",
+        "vitest-tsconfig-paths": "^3.4.1"
-		"vite-tsconfig-paths": "^5.1.4",
+    }
-		"vitest": "^3.1.2",
-		"vitest-tsconfig-paths": "^3.4.1"
-	}
 }

packages/backend/db/config.ts

@@ -0,0 +1,46 @@
+const requiredEnvVars = ["DB_HOST", "DB_NAME", "DB_USER", "DB_PASSWORD", "DB_PORT", "DB_NAME_CRED"];
+
+const unsetVars = requiredEnvVars.filter((key) => process.env[key] === undefined);
+
+if (unsetVars.length > 0) {
+	throw new Error(`Missing required environment variables: ${unsetVars.join(", ")}`);
+}
+
+const databaseHost = process.env["DB_HOST"]!;
+const databaseName = process.env["DB_NAME"];
+const databaseNameCred = process.env["DB_NAME_CRED"]!;
+const databaseUser = process.env["DB_USER"]!;
+const databasePassword = process.env["DB_PASSWORD"]!;
+const databasePort = process.env["DB_PORT"]!;
+
+export const postgresConfig = {
+	hostname: databaseHost,
+	port: parseInt(databasePort),
+	database: databaseName,
+	user: databaseUser,
+	password: databasePassword
+};
+
+export const postgresConfigNpm = {
+	host: databaseHost,
+	port: parseInt(databasePort),
+	database: databaseName,
+	username: databaseUser,
+	password: databasePassword
+};
+
+export const postgresCredConfigNpm = {
+	host: databaseHost,
+	port: parseInt(databasePort),
+	database: databaseNameCred,
+	username: databaseUser,
+	password: databasePassword
+}
+
+export const postgresConfigCred = {
+	hostname: databaseHost,
+	port: parseInt(databasePort),
+	database: databaseNameCred,
+	user: databaseUser,
+	password: databasePassword
+};

packages/backend/db/db.ts

@@ -0,0 +1,5 @@
+import postgres from "postgres";
+import { postgresConfigNpm, postgresCredConfigNpm } from "./config";
+
+export const sql = postgres(postgresConfigNpm);
+export const sqlCred = postgres(postgresCredConfigNpm)

packages/backend/db/latestSnapshots.ts

@@ -1,13 +0,0 @@
-import { sql } from "@core/db/dbNew";
-import type { LatestSnapshotType } from "@core/db/schema.d.ts";
-
-export async function getVideosInViewsRange(minViews: number, maxViews: number) {
-	return sql<LatestSnapshotType[]>`
-        SELECT *
-        FROM latest_video_snapshot
-        WHERE views >= ${minViews}
-        AND views <= ${maxViews}
-        ORDER BY views DESC
-        LIMIT 5000
-	`;
-}

packages/backend/db/videoInfo.ts

@@ -1,15 +1,15 @@
 import logger from "@core/log/logger.ts";
 import { redis } from "@core/db/redis.ts";
-import { sql } from "@core/db/dbNew.ts";
+import { sql } from "./db.ts";
 import { number, ValidationError } from "yup";
-import { createHandlers } from "@/src/utils.ts";
+import { createHandlers } from "../src/utils.ts";
 import { getVideoInfo, getVideoInfoByBV } from "@core/net/getVideoInfo.ts";
-import { idSchema } from "./snapshots.ts";
+import { idSchema } from "../routes/snapshots.ts";
 import { NetSchedulerError } from "@core/net/delegate.ts";
 import type { Context } from "hono";
 import type { BlankEnv, BlankInput } from "hono/types";
 import type { VideoInfoData } from "@core/net/bilibili.d.ts";
-import { startTime, endTime } from "hono/timing";
+import { startTime, endTime } from 'hono/timing'
 
 const CACHE_EXPIRATION_SECONDS = 60;
 
@@ -34,7 +34,7 @@ async function insertVideoSnapshot(data: VideoInfoData) {
 }
 
 export const videoInfoHandler = createHandlers(async (c: ContextType) => {
-	startTime(c, "parse", "Parse the request");
+	startTime(c, 'parse', 'Parse the request');
 	try {
 		const id = await idSchema.validate(c.req.param("id"));
 		let videoId: string | number = id as string;
@@ -45,33 +45,33 @@ export const videoInfoHandler = createHandlers(async (c: ContextType) => {
 		}
 
 		const cacheKey = `cvsa:videoInfo:${videoId}`;
-		endTime(c, "parse");
+		endTime(c, 'parse');
-		startTime(c, "cache", "Check for cached data");
+		startTime(c, 'cache', 'Check for cached data');
 		const cachedData = await redis.get(cacheKey);
-		endTime(c, "cache");
+		endTime(c, 'cache');
 		if (cachedData) {
 			return c.json(JSON.parse(cachedData));
 		}
-		startTime(c, "net", "Fetch data");
+		startTime(c, 'net', 'Fetch data');
 		let result: VideoInfoData | number;
 		if (typeof videoId === "number") {
 			result = await getVideoInfo(videoId, "getVideoInfo");
 		} else {
 			result = await getVideoInfoByBV(videoId, "getVideoInfo");
 		}
-		endTime(c, "net");
+		endTime(c, 'net');
 
 		if (typeof result === "number") {
 			return c.json({ message: "Error fetching video info", code: result }, 500);
 		}
 
-		startTime(c, "db", "Write data to database");
+		startTime(c, 'db', 'Write data to database');
 
 		await redis.setex(cacheKey, CACHE_EXPIRATION_SECONDS, JSON.stringify(result));
 
 		await insertVideoSnapshot(result);
 
-		endTime(c, "db");
+		endTime(c, 'db');
 		return c.json(result);
 	} catch (e) {
 		if (e instanceof ValidationError) {

packages/backend/db/videoSnapshot.ts

@@ -1,4 +1,4 @@
-import { sql } from "@core/db/dbNew";
+import { sql } from "./db";
 import type { VideoSnapshotType } from "@core/db/schema.d.ts";
 
 export async function getVideoSnapshots(

packages/backend/lib/auth/captchaDifficulty.ts

@@ -1,62 +0,0 @@
-import { Psql } from "@core/db/psql";
-import { SlidingWindow } from "@core/mq/slidingWindow.ts";
-import { redis } from "@core/db/redis.ts";
-import { getIdentifier } from "@/middleware/rateLimiters.ts";
-import { Context } from "hono";
-
-type seconds = number;
-
-export interface CaptchaDifficultyConfig {
-	global: boolean;
-	duration: seconds;
-	threshold: number;
-	difficulty: number;
-}
-
-export const getCaptchaDifficultyConfigByRoute = async (sql: Psql, route: string): Promise<CaptchaDifficultyConfig[]> =>  {
-	return sql<CaptchaDifficultyConfig[]>`
-		SELECT duration, threshold, difficulty, global
-		FROM captcha_difficulty_settings
-		WHERE CONCAT(method, '-', path) = ${route}
-		ORDER BY duration
-	`;
-};
-
-export const getCaptchaConfigMaxDuration = async (sql: Psql, route: string): Promise<seconds> => {
-	const rows = await sql<{max: number}[]>`
-		SELECT MAX(duration)
-		FROM captcha_difficulty_settings
-		WHERE CONCAT(method, '-', path) = ${route}
-	`;
-	if (rows.length < 1){
-		return Number.MAX_SAFE_INTEGER;
-	}
-	return rows[0].max;
-}
-
-
-export const getCurrentCaptchaDifficulty = async (sql: Psql, c: Context | string): Promise<number | null> => {
-	const isRoute = typeof c === "string";
-	const route = isRoute ? c : `${c.req.method}-${c.req.path}`
-	const configs = await getCaptchaDifficultyConfigByRoute(sql, route);
-	if (configs.length < 1) {
-		return null
-	}
-	else if (configs.length == 1) {
-		return configs[0].difficulty
-	}
-	const maxDuration = configs.reduce((max, config) =>
-		Math.max(max, config.duration), 0);
-	const slidingWindow = new SlidingWindow(redis, maxDuration);
-	for (let i = 1; i < configs.length; i++) {
-		const config = configs[i];
-		const lastConfig = configs[i - 1];
-		const identifier = isRoute ? c : getIdentifier(c, config.global);
-		const count = await slidingWindow.count(`captcha-${identifier}`, config.duration);
-		if (count >= config.threshold) {
-			continue;
-		}
-		return lastConfig.difficulty
-	}
-	return configs[configs.length-1].difficulty;
-}

packages/backend/lib/auth/getJWTsecret.ts

@@ -1,14 +0,0 @@
-import { ErrorResponse } from "src/schema";
-
-export const getJWTsecret = () => {
-	const secret = process.env["JWT_SECRET"];
-	if (!secret) {
-		const response: ErrorResponse = {
-			message: "JWT_SECRET is not set",
-			code: "SERVER_ERROR",
-			errors: []
-		};
-		return [response, true];
-	}
-	return [secret, null];
-};

packages/backend/middleware/captcha.ts

@@ -1,120 +0,0 @@
-import { Context, Next } from "hono";
-import { ErrorResponse } from "src/schema";
-import { SlidingWindow } from "@core/mq/slidingWindow.ts";
-import { getCaptchaConfigMaxDuration, getCurrentCaptchaDifficulty } from "@/lib/auth/captchaDifficulty.ts";
-import { sqlCred } from "@core/db/dbNew.ts";
-import { redis } from "@core/db/redis.ts";
-import { verify } from "hono/jwt";
-import { JwtTokenInvalid, JwtTokenExpired } from "hono/utils/jwt/types";
-import { getJWTsecret } from "@/lib/auth/getJWTsecret.ts";
-import { lockManager } from "@core/mq/lockManager.ts";
-import { object, string, number, ValidationError } from "yup";
-import { getIdentifier } from "@/middleware/rateLimiters.ts";
-
-const tokenSchema = object({
-	exp: number().integer(),
-	id: string().length(6),
-	difficulty: number().integer().moreThan(0)
-});
-
-export const captchaMiddleware = async (c: Context, next: Next) => {
-	const authHeader = c.req.header("Authorization");
-
-	if (!authHeader) {
-		const response: ErrorResponse = {
-			message: "'Authorization' header is missing.",
-			code: "UNAUTHORIZED",
-			errors: []
-		};
-		return c.json<ErrorResponse>(response, 401);
-	}
-
-	const authIsBearer = authHeader.startsWith("Bearer ");
-	if (!authIsBearer || authHeader.length < 8) {
-		const response: ErrorResponse = {
-			message: "'Authorization' header is invalid.",
-			code: "INVALID_HEADER",
-			errors: []
-		};
-		return c.json<ErrorResponse>(response, 400);
-	}
-
-	const [r, err] = getJWTsecret();
-	if (err) {
-		return c.json<ErrorResponse>(r as ErrorResponse, 500);
-	}
-	const jwtSecret = r as string;
-
-	const token = authHeader.substring(7);
-
-	const path = c.req.path;
-	const method = c.req.method;
-	const route = `${method}-${path}`;
-
-	const requiredDifficulty = await getCurrentCaptchaDifficulty(sqlCred, c);
-
-	try {
-		const decodedPayload = await verify(token, jwtSecret);
-		const payload = await tokenSchema.validate(decodedPayload);
-		const difficulty = payload.difficulty;
-		const tokenID = payload.id;
-		const consumed = await lockManager.isLocked(tokenID);
-		if (consumed) {
-			const response: ErrorResponse = {
-				message: "Token has already been used.",
-				code: "INVALID_CREDENTIALS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 401);
-		}
-		if (difficulty < requiredDifficulty) {
-			const response: ErrorResponse = {
-				message: "Token too weak.",
-				code: "UNAUTHORIZED",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 401);
-		}
-		const EXPIRE_FIVE_MINUTES = 300;
-		await lockManager.acquireLock(tokenID, EXPIRE_FIVE_MINUTES);
-	} catch (e) {
-		if (e instanceof JwtTokenInvalid) {
-			const response: ErrorResponse = {
-				message: "Failed to verify the token.",
-				code: "INVALID_CREDENTIALS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else if (e instanceof JwtTokenExpired) {
-			const response: ErrorResponse = {
-				message: "Token expired.",
-				code: "INVALID_CREDENTIALS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else if (e instanceof ValidationError) {
-			const response: ErrorResponse = {
-				code: "INVALID_QUERY_PARAMS",
-				message: "Invalid query parameters",
-				errors: e.errors
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else {
-			const response: ErrorResponse = {
-				message: "Unknown error.",
-				code: "UNKNOWN_ERROR",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 500);
-		}
-	}
-	const duration = await getCaptchaConfigMaxDuration(sqlCred, route);
-	const window = new SlidingWindow(redis, duration);
-
-	const identifierWithIP = getIdentifier(c, true);
-	const identifier = getIdentifier(c, false);
-	await window.event(`captcha-${identifier}`);
-	await window.event(`captcha-${identifierWithIP}`);
-
-	await next();
-};

packages/backend/middleware/contentType.ts

@@ -3,4 +3,4 @@ import { Context, Next } from "hono";
 export const contentType = async (c: Context, next: Next) => {
 	await next();
 	c.header("Content-Type", "application/json; charset=utf-8");
-};
+};

packages/backend/middleware/cors.ts

@@ -1,14 +0,0 @@
-import { cors } from "hono/cors";
-import { Context, Next } from "hono";
-
-export const corsMiddleware = async (c: Context, next: Next) => {
-	if (c.req.path.startsWith("/user") || c.req.path.startsWith("/login")) {
-		const corsMiddlewareHandler = cors({
-			origin: c.req.header("Origin"),
-			credentials: true
-		});
-		return corsMiddlewareHandler(c, next);
-	}
-	const corsMiddlewareHandler = cors();
-	return corsMiddlewareHandler(c, next);
-};

packages/backend/middleware/index.ts

@@ -0,0 +1,19 @@
+import { Hono } from "hono";
+import { Variables } from "hono/types";
+import { bodyLimitForPing } from "./bodyLimits.ts";
+import { pingHandler } from "../routes/ping.ts";
+import { registerRateLimiter } from "./rateLimiters.ts";
+import { preetifyResponse } from "./preetifyResponse.ts";
+import { logger } from "./logger.ts";
+import { timing } from "hono/timing";
+import { contentType } from "./contentType.ts";
+
+export function configureMiddleWares(app: Hono<{Variables: Variables }>) {
+	app.use("*", contentType);
+	app.use(timing());
+	app.use("*", preetifyResponse);
+	app.use("*", logger({}));
+
+	app.post("/user", registerRateLimiter);
+	app.all("/ping", bodyLimitForPing, ...pingHandler);
+}

packages/backend/middleware/logger.ts

@@ -77,7 +77,7 @@ const defaultFormatter = (params) => {
 		`${methodColor} ${params.method.padEnd(6)}${reset} ${params.path}`
 	);
 };
-type Ctx = Context;
+type Ctx = Context
 export const logger = (config) => {
 	const { formatter = defaultFormatter, output = console, skipPaths = [], skip = null } = config;
 

packages/backend/middleware/preetifyResponse.ts

@@ -15,4 +15,4 @@ export const preetifyResponse = async (c: Context, next: Next) => {
 		endTime(c, "seralize");
 		c.res = new Response(prettyJson, { headers: { "Content-Type": "text/plain; charset=utf-8" } });
 	}
-};
+};

packages/backend/middleware/rateLimiters.ts

@@ -1,53 +1,27 @@
+import { rateLimiter, Store } from "hono-rate-limiter";
 import type { BlankEnv } from "hono/types";
+import { MINUTE } from "@core/const/time.ts";
 import { getConnInfo } from "hono/bun";
-import { Context, Next } from "hono";
+import type { Context } from "hono";
-import { generateRandomId } from "@core/lib/randomID.ts";
-import { RateLimiter } from "@koshnic/ratelimit";
-import { ErrorResponse } from "@/src/schema";
 import { redis } from "@core/db/redis.ts";
+import { RedisStore } from "rate-limit-redis";
 
-export const getUserIP = (c: Context) => {
+export const registerRateLimiter = rateLimiter<BlankEnv, "/user", {}>({
-	let ipAddr = null;
+	windowMs: 60 * MINUTE,
-	const info = getConnInfo(c);
+	limit: 10,
-	if (info.remote && info.remote.address) {
+	standardHeaders: "draft-6",
-		ipAddr = info.remote.address;
+	keyGenerator: (c) => {
-	}
+		const info = getConnInfo(c as unknown as Context<BlankEnv, "/user", {}>);
-	const forwardedFor = c.req.header("X-Forwarded-For");
+		if (!info.remote || !info.remote.address) {
-	if (forwardedFor) {
+			return crypto.randomUUID();
-		ipAddr = forwardedFor.split(",")[0];
+		}
-	}
+		const addr = info.remote.address;
-	return ipAddr;
+		const path = new URL(c.req.url).pathname;
-};
+		const method = c.req.method;
-
+		return `${method}-${path}@${addr}`;
-export const getIdentifier = (c: Context, includeIP: boolean = true) => {
+	},
-	let ipAddr = generateRandomId(6);
+	store: new RedisStore({
-	if (getUserIP(c)) {
+		// @ts-expect-error - Known issue: the `c`all` function is not present in @types/ioredis
-		ipAddr = getUserIP(c);
+		sendCommand: (...args: string[]) => redis.call(...args)
-	}
+	}) as unknown as Store
-	const path = c.req.path;
+});
-	const method = c.req.method;
-	const ipIdentifier = includeIP ? `@${ipAddr}` : "";
-	return `${method}-${path}${ipIdentifier}`;
-};
-
-export const registerRateLimiter = async (c: Context<BlankEnv, "/user", {}>, next: Next) => {
-	const limiter = new RateLimiter(redis);
-	const identifier = getIdentifier(c, true);
-	const { allowed, retryAfter } = await limiter.allow(identifier, {
-		burst: 5,
-		ratePerPeriod: 5,
-		period: 120,
-		cost: 1
-	});
-
-	if (!allowed) {
-		const response: ErrorResponse = {
-			message: `Too many requests, please retry after ${Math.round(retryAfter)} seconds.`,
-			code: "RATE_LIMIT_EXCEEDED",
-			errors: []
-		};
-		return c.json<ErrorResponse>(response, 429);
-	}
-
-	await next();
-};

packages/backend/package.json

@@ -1,21 +1,15 @@
 {
-	"name": "@cvsa/backend",
+	"name": "backend",
-	"private": false,
-	"version": "0.6.0",
 	"scripts": {
 		"format": "prettier --write .",
 		"dev": "NODE_ENV=development bun run --hot src/main.ts",
-		"start": "NODE_ENV=production bun run src/main.ts",
+		"start": "NODE_ENV=production bun run src/main.ts"
-		"build": "bun build ./src/main.ts --target bun --outdir ./dist"
 	},
 	"dependencies": {
-		"@koshnic/ratelimit": "^1.0.3",
 		"@rabbit-company/argon2id": "^2.1.0",
-		"chalk": "^5.4.1",
 		"hono": "^4.7.8",
 		"hono-rate-limiter": "^0.4.2",
 		"ioredis": "^5.6.1",
-		"limiter": "^3.0.0",
 		"postgres": "^3.4.5",
 		"rate-limit-redis": "^4.2.0",
 		"yup": "^1.6.1",
@@ -24,7 +18,5 @@
 	"devDependencies": {
 		"@types/bun": "^1.2.11",
 		"prettier": "^3.5.3"
-	},
+	}
-	"main": "./dist/main.js",
-	"types": "./src/types.d.ts"
 }

packages/backend/routes/404.ts

@@ -7,4 +7,4 @@ export const notFoundRoute = (c: Context) => {
 		},
 		404
 	);
-};
+};

packages/backend/routes/captcha/[id]/result/GET.ts

@@ -1,99 +0,0 @@
-import { Context } from "hono";
-import { Bindings, BlankEnv } from "hono/types";
-import { ErrorResponse } from "src/schema";
-import { createHandlers } from "src/utils.ts";
-import { sign } from "hono/jwt";
-import { generateRandomId } from "@core/lib/randomID.ts";
-import { getJWTsecret } from "lib/auth/getJWTsecret.ts";
-
-interface CaptchaResponse {
-	success: boolean;
-	difficulty?: number;
-	error?: string;
-}
-
-const getChallengeVerificationResult = async (id: string, ans: string) => {
-	const baseURL = process.env["UCAPTCHA_URL"];
-	const url = new URL(baseURL);
-	url.pathname = `/challenge/${id}/validation`;
-	return await fetch(url.toString(), {
-		method: "POST",
-		headers: {
-			"Content-Type": "application/json"
-		},
-		body: JSON.stringify({
-			y: ans
-		})
-	});
-};
-
-export const verifyChallengeHandler = createHandlers(
-	async (c: Context<BlankEnv & { Bindings: Bindings }, "/captcha/:id/result">) => {
-		const id = c.req.param("id");
-		const ans = c.req.query("ans");
-		if (!ans) {
-			const response: ErrorResponse = {
-				message: "Missing required query parameter: ans",
-				code: "INVALID_QUERY_PARAMS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 400);
-		}
-		const res = await getChallengeVerificationResult(id, ans);
-		const data: CaptchaResponse = await res.json();
-		if (data.error && res.status === 404) {
-			const response: ErrorResponse = {
-				message: data.error,
-				code: "ENTITY_NOT_FOUND",
-				i18n: {
-					key: "backend.error.captcha_not_found"
-				},
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 401);
-		} else if (data.error && res.status === 400) {
-			const response: ErrorResponse = {
-				message: data.error,
-				code: "INVALID_QUERY_PARAMS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else if (data.error) {
-			const response: ErrorResponse = {
-				message: data.error,
-				code: "UNKNOWN_ERROR",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 500);
-		}
-		if (!data.success) {
-			const response: ErrorResponse = {
-				message: "Incorrect answer",
-				code: "INVALID_CREDENTIALS",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 401);
-		}
-
-		const [r, err] = getJWTsecret();
-		if (err) {
-			return c.json<ErrorResponse>(r as ErrorResponse, 500);
-		}
-		const jwtSecret = r as string;
-
-		const tokenID = generateRandomId(6);
-		const NOW = Math.floor(Date.now() / 1000);
-		const FIVE_MINUTES_LATER = NOW + 60 * 5;
-		const jwt = await sign(
-			{
-				difficulty: data.difficulty!,
-				id: tokenID,
-				exp: FIVE_MINUTES_LATER
-			},
-			jwtSecret
-		);
-		return c.json({
-			token: jwt
-		});
-	}
-);

packages/backend/routes/captcha/difficulty/GET.ts

@@ -1,44 +0,0 @@
-import { createHandlers } from "src/utils.ts";
-import { object, string, ValidationError } from "yup";
-import { ErrorResponse } from "src/schema";
-import { getCurrentCaptchaDifficulty } from "@/lib/auth/captchaDifficulty.ts";
-import { sqlCred } from "@core/db/dbNew.ts";
-
-const queryParamsSchema = object({
-	route: string().matches(/(?:GET|POST|PUT|PATCH|DELETE)-\/.*/g)
-});
-
-export const getCaptchaDifficultyHandler = createHandlers(async (c) => {
-	try {
-		const queryParams = await queryParamsSchema.validate(c.req.query());
-		const { route } = queryParams;
-		const difficulty = await getCurrentCaptchaDifficulty(sqlCred, route);
-		if (!difficulty) {
-			const response: ErrorResponse<unknown> = {
-				code: "ENTITY_NOT_FOUND",
-				message: "No difficulty configs found for this route.",
-				errors: []
-			};
-			return c.json<ErrorResponse<unknown>>(response, 404);
-		}
-		return c.json({
-			difficulty: difficulty
-		});
-	} catch (e: unknown) {
-		if (e instanceof ValidationError) {
-			const response: ErrorResponse = {
-				code: "INVALID_QUERY_PARAMS",
-				message: "Invalid query parameters",
-				errors: e.errors
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else {
-			const response: ErrorResponse<unknown> = {
-				code: "UNKNOWN_ERROR",
-				message: "Unknown error",
-				errors: [e]
-			};
-			return c.json<ErrorResponse<unknown>>(response, 500);
-		}
-	}
-});

packages/backend/routes/captcha/index.ts

@@ -1,2 +0,0 @@
-export * from "./session/POST.ts";
-export * from "./[id]/result/GET.ts";

packages/backend/routes/captcha/session/POST.ts

@@ -1,51 +0,0 @@
-import { createHandlers } from "src/utils.ts";
-import { getCurrentCaptchaDifficulty } from "@/lib/auth/captchaDifficulty.ts";
-import { sqlCred } from "@core/db/dbNew.ts";
-import { object, string, ValidationError } from "yup";
-import { CaptchaSessionResponse, ErrorResponse } from "@/src/schema";
-import type { ContentfulStatusCode } from "hono/utils/http-status";
-
-const bodySchema = object({
-	route: string().matches(/(?:GET|POST|PUT|PATCH|DELETE)-\/.*/g)
-});
-
-const createNewChallenge = async (difficulty: number) => {
-	const baseURL = process.env["UCAPTCHA_URL"];
-	const url = new URL(baseURL);
-	url.pathname = "/challenge";
-	return await fetch(url.toString(), {
-		method: "POST",
-		headers: {
-			"Content-Type": "application/json"
-		},
-		body: JSON.stringify({
-			difficulty: difficulty
-		})
-	});
-};
-
-export const createCaptchaSessionHandler = createHandlers(async (c) => {
-	try {
-		const requestBody = await bodySchema.validate(await c.req.json());
-		const { route } = requestBody;
-		const difficuly = await getCurrentCaptchaDifficulty(sqlCred, route);
-		const res = await createNewChallenge(difficuly);
-		return c.json<CaptchaSessionResponse | unknown>(await res.json(), res.status as ContentfulStatusCode);
-	} catch (e: unknown) {
-		if (e instanceof ValidationError) {
-			const response: ErrorResponse = {
-				code: "INVALID_QUERY_PARAMS",
-				message: "Invalid query parameters",
-				errors: e.errors
-			};
-			return c.json<ErrorResponse>(response, 400);
-		} else {
-			const response: ErrorResponse<unknown> = {
-				code: "UNKNOWN_ERROR",
-				message: "Unknown error",
-				errors: [e]
-			};
-			return c.json<ErrorResponse<unknown>>(response, 500);
-		}
-	}
-});

packages/backend/routes/index.ts

@@ -1,29 +1,17 @@
-import { getSingerForBirthday, pickSinger, pickSpecialSinger, type Singer } from "lib/const/singers.ts";
+import { rootHandler } from "./root/root.ts";
-import { VERSION } from "src/main.ts";
+import { pingHandler } from "./ping.ts";
-import { createHandlers } from "src/utils.ts";
+import { getSnapshotsHanlder } from "./snapshots.ts";
+import { registerHandler } from "./user.ts";
+import { videoInfoHandler } from "db/videoInfo.ts";
+import { Hono } from "hono";
+import { Variables } from "hono/types";
 
-export const rootHandler = createHandlers((c) => {
+export function configureRoutes(app: Hono<{Variables: Variables }>) {
-	let singer: Singer | Singer[];
+	app.get("/", ...rootHandler);
-	const shouldShowSpecialSinger = Math.random() < 0.016;
+	app.all("/ping", ...pingHandler);
-	if (getSingerForBirthday().length !== 0) {
+
-		singer = JSON.parse(JSON.stringify(getSingerForBirthday())) as Singer[];
+	app.get("/video/:id/snapshots", ...getSnapshotsHanlder);
-		for (const s of singer) {
+	app.post("/user", ...registerHandler);
-			delete s.birthday;
+
-			s.message = `祝${s.name}生日快乐~`;
+	app.get("/video/:id/info", ...videoInfoHandler);
-		}
+}
-	} else if (shouldShowSpecialSinger) {
-		singer = pickSpecialSinger();
-	} else {
-		singer = pickSinger();
-	}
-	return c.json({
-		project: {
-			name: "中V档案馆",
-			motto: "一起唱吧，心中的歌！"
-		},
-		status: 200,
-		version: VERSION,
-		time: Date.now(),
-		singer: singer
-	});
-});

packages/backend/routes/login/session/POST.ts

@@ -1,105 +0,0 @@
-import { Context } from "hono";
-import { Bindings, BlankEnv } from "hono/types";
-import { ErrorResponse, LoginResponse } from "src/schema";
-import { createHandlers } from "src/utils.ts";
-import { sqlCred } from "@core/db/dbNew";
-import { object, string, ValidationError } from "yup";
-import { setCookie } from "hono/cookie";
-import Argon2id from "@rabbit-company/argon2id";
-import { createLoginSession } from "routes/user/POST";
-import { UserType } from "@core/db/schema";
-
-const LoginBodySchema = object({
-	username: string().trim().required("Username is required").max(50, "Username cannot exceed 50 characters"),
-	password: string().required("Password is required")
-});
-
-export const loginHandler = createHandlers(
-	async (c: Context<BlankEnv & { Bindings: Bindings }, "/user/session/:id">) => {
-		try {
-			const body = await LoginBodySchema.validate(await c.req.json());
-			const { username, password: submittedPassword } = body;
-
-			const result = await sqlCred<UserType[]>`
-			    SELECT *
-				FROM users
-				WHERE username = ${username}
-			`;
-
-			if (result.length === 0) {
-				const response: ErrorResponse<string> = {
-					message: `User does not exist.`,
-					errors: [`User ${username} does not exist.`],
-					code: "ENTITY_NOT_FOUND"
-				};
-				return c.json<ErrorResponse<string>>(response, 400);
-			}
-
-			const storedPassword = result[0].password;
-			const uid = result[0].id;
-			const nickname = result[0].nickname;
-			const role = result[0].role;
-
-			const passwordAreSame = await Argon2id.verify(storedPassword, submittedPassword);
-
-			if (!passwordAreSame) {
-				const response: ErrorResponse<string> = {
-					message: "Incorrect password.",
-					errors: [],
-					i18n: {
-						key: "backend.error.incorrect_password"
-					},
-					code: "INVALID_CREDENTIALS"
-				};
-				return c.json<ErrorResponse<string>>(response, 401);
-			}
-
-			const sessionID = await createLoginSession(uid, c);
-
-			const response: LoginResponse = {
-				uid: uid,
-				username: username,
-				nickname: nickname,
-				role: role,
-				token: sessionID
-			};
-
-			const A_YEAR = 365 * 86400;
-			const isDev = process.env.NODE_ENV === "development";
-
-			setCookie(c, "session_id", sessionID, {
-				path: "/",
-				maxAge: A_YEAR,
-				domain: process.env.DOMAIN,
-				secure: isDev ? true : true,
-				sameSite: isDev ? "None" : "Lax",
-				httpOnly: true
-			});
-
-			return c.json<LoginResponse>(response, 200);
-		} catch (e) {
-			if (e instanceof ValidationError) {
-				const response: ErrorResponse<string> = {
-					message: "Invalid registration data.",
-					errors: e.errors,
-					code: "INVALID_PAYLOAD"
-				};
-				return c.json<ErrorResponse<string>>(response, 400);
-			} else if (e instanceof SyntaxError) {
-				const response: ErrorResponse<string> = {
-					message: "Invalid JSON payload.",
-					errors: [e.message],
-					code: "INVALID_FORMAT"
-				};
-				return c.json<ErrorResponse<string>>(response, 400);
-			} else {
-				const response: ErrorResponse<string> = {
-					message: "Unknown error.",
-					errors: [(e as Error).message],
-					code: "UNKNOWN_ERROR"
-				};
-				return c.json<ErrorResponse<string>>(response, 500);
-			}
-		}
-	}
-);

packages/backend/routes/ping.ts

@@ -0,0 +1,24 @@
+import { getClientIP } from "middleware/logger.ts";
+import { createHandlers } from "../src/utils.ts";
+import { VERSION } from "../src/main.ts";
+
+export const pingHandler = createHandlers(async (c) => {
+	const requestHeaders = c.req.raw.headers;
+	return c.json({
+		"message": "pong",
+		"request": {
+			"headers": requestHeaders,
+			"ip": getClientIP(c),
+			"mode": c.req.raw.mode,
+			"method": c.req.method,
+			"query": new URL(c.req.url).searchParams,
+			"body": await c.req.text(),
+			"url": c.req.raw.url
+		},
+		"response": {
+			"time": new Date().getTime(),
+			"status": 200,
+			"version": VERSION,
+		}
+	});
+});

packages/backend/routes/ping/index.ts

@@ -1,24 +0,0 @@
-import { getClientIP } from "middleware/logger.ts";
-import { createHandlers } from "src/utils.ts";
-import { VERSION } from "src/main.ts";
-
-export const pingHandler = createHandlers(async (c) => {
-	const requestHeaders = c.req.raw.headers;
-	return c.json({
-		message: "pong",
-		request: {
-			headers: requestHeaders,
-			ip: getClientIP(c),
-			mode: c.req.raw.mode,
-			method: c.req.method,
-			query: new URL(c.req.url).searchParams,
-			body: await c.req.text(),
-			url: c.req.raw.url
-		},
-		response: {
-			time: new Date().getTime(),
-			status: 200,
-			version: VERSION
-		}
-	});
-});

packages/backend/routes/root/root.ts

@@ -0,0 +1,29 @@
+import { getSingerForBirthday, pickSinger, pickSpecialSinger, type Singer } from "./singers.ts";
+import { VERSION } from "../../src/main.ts";
+import { createHandlers } from "../../src/utils.ts";
+
+export const rootHandler = createHandlers((c) => {
+	let singer: Singer | Singer[];
+	const shouldShowSpecialSinger = Math.random() < 0.016;
+	if (getSingerForBirthday().length !== 0) {
+		singer = JSON.parse(JSON.stringify(getSingerForBirthday())) as Singer[];
+		for (const s of singer) {
+			delete s.birthday;
+			s.message = `祝${s.name}生日快乐~`;
+		}
+	} else if (shouldShowSpecialSinger) {
+		singer = pickSpecialSinger();
+	} else {
+		singer = pickSinger();
+	}
+	return c.json({
+		project: {
+			name: "中V档案馆",
+			motto: "一起唱吧，心中的歌！"
+		},
+		status: 200,
+		version: VERSION,
+		time: Date.now(),
+		singer: singer
+	});
+});

packages/backend/routes/session/[id]/DELETE.ts

@@ -1,75 +0,0 @@
-import { Context } from "hono";
-import { Bindings, BlankEnv } from "hono/types";
-import { ErrorResponse } from "src/schema";
-import { createHandlers } from "src/utils.ts";
-import { sqlCred } from "@core/db/dbNew";
-import { object, string, ValidationError } from "yup";
-import { setCookie } from "hono/cookie";
-
-const loginSessionExists = async (sessionID: string) => {
-	const result = await sqlCred`
-        SELECT 1
-        FROM login_sessions
-        WHERE id = ${sessionID}
-	`;
-	return result.length > 0;
-};
-
-export const logoutHandler = createHandlers(async (c: Context<BlankEnv & { Bindings: Bindings }, "/session/:id">) => {
-	try {
-		const session_id = c.req.param("id");
-
-		const exists = loginSessionExists(session_id);
-
-		if (!exists) {
-			const response: ErrorResponse<string> = {
-				message: "Cannot found given session_id.",
-				errors: [`Session ${session_id} not found`],
-				code: "ENTITY_NOT_FOUND"
-			};
-			return c.json<ErrorResponse<string>>(response, 404);
-		}
-
-		await sqlCred`
-		    UPDATE login_sessions
-			SET deactivated_at = CURRENT_TIMESTAMP
-			WHERE id = ${session_id}
-		`;
-
-		const isDev = process.env.NODE_ENV === "development";
-
-		setCookie(c, "session_id", "", {
-			path: "/",
-			maxAge: 0,
-			domain: process.env.DOMAIN,
-			secure: isDev ? true : true,
-			sameSite: isDev ? "None" : "Lax",
-			httpOnly: true
-		});
-
-		return c.body(null, 204);
-	} catch (e) {
-		if (e instanceof ValidationError) {
-			const response: ErrorResponse<string> = {
-				message: "Invalid registration data.",
-				errors: e.errors,
-				code: "INVALID_PAYLOAD"
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		} else if (e instanceof SyntaxError) {
-			const response: ErrorResponse<string> = {
-				message: "Invalid JSON payload.",
-				errors: [e.message],
-				code: "INVALID_FORMAT"
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		} else {
-			const response: ErrorResponse<string> = {
-				message: "Unknown error.",
-				errors: [(e as Error).message],
-				code: "UNKNOWN_ERROR"
-			};
-			return c.json<ErrorResponse<string>>(response, 500);
-		}
-	}
-});

packages/backend/routes/session/index.ts

@@ -1 +0,0 @@
-export * from "./[id]/DELETE";

packages/backend/routes/snapshots.ts

@@ -1,10 +1,10 @@
 import type { Context } from "hono";
-import { createHandlers } from "src/utils.ts";
+import { createHandlers } from "../src/utils.ts";
 import type { BlankEnv, BlankInput } from "hono/types";
-import { getVideoSnapshots, getVideoSnapshotsByBV } from "db/snapshots.ts";
+import { getVideoSnapshots, getVideoSnapshotsByBV } from "../db/videoSnapshot.ts";
 import type { VideoSnapshotType } from "@core/db/schema.d.ts";
 import { boolean, mixed, number, object, ValidationError } from "yup";
-import { ErrorResponse } from "src/schema";
+import { ErrorResponse } from "../src/schema";
 import { startTime, endTime } from "hono/timing";
 
 const SnapshotQueryParamsSchema = object({
@@ -96,7 +96,7 @@ export const getSnapshotsHanlder = createHandlers(async (c: ContextType) => {
 			return c.json<ErrorResponse<string>>(response, 400);
 		} else {
 			const response: ErrorResponse<unknown> = {
-				code: "UNKNOWN_ERROR",
+				code: "UNKNOWN_ERR",
 				message: "Unhandled error",
 				errors: [e]
 			};

packages/backend/routes/user.ts

@@ -0,0 +1,74 @@
+import { createHandlers } from "src/utils.ts";
+import Argon2id from "@rabbit-company/argon2id";
+import { object, string, ValidationError } from "yup";
+import type { Context } from "hono";
+import type { Bindings, BlankEnv, BlankInput } from "hono/types";
+import { sqlCred } from "db/db.ts";
+import { ErrorResponse, StatusResponse } from "src/schema";
+
+const RegistrationBodySchema = object({
+	username: string().trim().required("Username is required").max(50, "Username cannot exceed 50 characters"),
+	password: string().required("Password is required"),
+	nickname: string().optional()
+});
+
+type ContextType = Context<BlankEnv & { Bindings: Bindings }, "/user", BlankInput>;
+
+export const userExists = async (username: string) => {
+	const result = await sqlCred`
+        SELECT 1
+        FROM users
+        WHERE username = ${username}
+	`;
+	return result.length > 0;
+};
+
+export const registerHandler = createHandlers(async (c: ContextType) => {
+	try {
+		const body = await RegistrationBodySchema.validate(await c.req.json());
+		const { username, password, nickname } = body;
+
+		if (await userExists(username)) {
+			const response: StatusResponse = {
+				message: `User "${username}" already exists.`
+			};
+			return c.json<StatusResponse>(response, 400);
+		}
+
+		const hash = await Argon2id.hashEncoded(password);
+
+		await sqlCred`
+            INSERT INTO users (username, password, nickname)
+            VALUES (${username}, ${hash}, ${nickname ? nickname : null})
+		`;
+
+		const response: StatusResponse = {
+			message: `User '${username}' registered successfully.`
+		}
+
+		return c.json<StatusResponse>(response, 201);
+	} catch (e) {
+		if (e instanceof ValidationError) {
+			const response: ErrorResponse<string> = {
+				message: "Invalid registration data.",
+				errors: e.errors,
+				code: "INVALID_PAYLOAD"
+			}
+			return c.json<ErrorResponse<string>>(response, 400);
+		} else if (e instanceof SyntaxError) {
+			const response: ErrorResponse<string> = {
+				message: "Invalid JSON payload.",
+				errors: [e.message],
+				code: "INVALID_FORMAT"
+			}
+			return c.json<ErrorResponse<string>>(response, 400);
+		} else {
+			const response: ErrorResponse<string> = {
+				message: "Invalid JSON payload.",
+				errors: [(e as Error).message],
+				code: "UNKNOWN_ERR"
+			}
+			return c.json<ErrorResponse<string>>(response, 500);
+		}
+	}
+});

packages/backend/routes/user/POST.ts

@@ -1,140 +0,0 @@
-import { createHandlers } from "src/utils.ts";
-import Argon2id from "@rabbit-company/argon2id";
-import { object, string, ValidationError } from "yup";
-import type { Context } from "hono";
-import type { Bindings, BlankEnv, BlankInput } from "hono/types";
-import { sqlCred } from "@core/db/dbNew.ts";
-import { ErrorResponse, SignUpResponse } from "src/schema";
-import { generateRandomId } from "@core/lib/randomID";
-import { getUserIP } from "@/middleware/rateLimiters";
-import { setCookie } from "hono/cookie";
-
-const RegistrationBodySchema = object({
-	username: string().trim().required("Username is required").max(50, "Username cannot exceed 50 characters"),
-	password: string().required("Password is required"),
-	nickname: string().optional()
-});
-
-type ContextType = Context<BlankEnv & { Bindings: Bindings }, "/user", BlankInput>;
-
-export const userExists = async (username: string) => {
-	const result = await sqlCred`
-        SELECT 1
-        FROM users
-        WHERE username = ${username}
-	`;
-	return result.length > 0;
-};
-
-export const createLoginSession = async (uid: number, c: Context): Promise<string> => {
-	const ipAddress = getUserIP(c) || null;
-	const userAgent = c.req.header("User-Agent") || null;
-	const id = generateRandomId(24);
-	await sqlCred`
-        INSERT INTO login_sessions (id, uid, expire_at, ip_address, user_agent)
-        VALUES (${id}, ${uid}, CURRENT_TIMESTAMP + INTERVAL '1 year', ${ipAddress}, ${userAgent})
-    `;
-	return id;
-};
-
-const getUserIDByName = async (username: string) => {
-	const result = await sqlCred<{ id: number }[]>`
-        SELECT id
-        FROM users
-        WHERE username = ${username}
-    `;
-	if (result.length === 0) {
-		return null;
-	}
-	return result[0].id;
-};
-
-export const registerHandler = createHandlers(async (c: ContextType) => {
-	try {
-		const body = await RegistrationBodySchema.validate(await c.req.json());
-		const { username, password, nickname } = body;
-
-		if (await userExists(username)) {
-			const response: ErrorResponse = {
-				message: `User "${username}" already exists.`,
-				code: "ENTITY_EXISTS",
-				errors: [],
-				i18n: {
-					key: "backend.error.user_exists",
-					values: {
-						username: username
-					}
-				}
-			};
-			return c.json<ErrorResponse>(response, 400);
-		}
-
-		const hash = await Argon2id.hashEncoded(password);
-
-		await sqlCred`
-            INSERT INTO users (username, password, nickname)
-            VALUES (${username}, ${hash}, ${nickname ? nickname : null})
-		`;
-
-		const uid = await getUserIDByName(username);
-
-		if (!uid) {
-			const response: ErrorResponse<string> = {
-				message: "Cannot find registered user.",
-				errors: [`Cannot find user ${username} in table 'users'.`],
-				code: "ENTITY_NOT_FOUND",
-				i18n: {
-					key: "backend.error.user_not_found_after_register",
-					values: {
-						username: username
-					}
-				}
-			};
-			return c.json<ErrorResponse<string>>(response, 500);
-		}
-
-		const sessionID = await createLoginSession(uid, c);
-
-		const response: SignUpResponse = {
-			username: username,
-			token: sessionID
-		};
-
-		const A_YEAR = 365 * 86400;
-		const isDev = process.env.NODE_ENV === "development";
-
-		setCookie(c, "session_id", sessionID, {
-			path: "/",
-			maxAge: A_YEAR,
-			domain: process.env.DOMAIN,
-			secure: isDev ? false : true,
-			sameSite: "Lax",
-			httpOnly: true
-		});
-
-		return c.json<SignUpResponse>(response, 201);
-	} catch (e) {
-		if (e instanceof ValidationError) {
-			const response: ErrorResponse<string> = {
-				message: "Invalid registration data.",
-				errors: e.errors,
-				code: "INVALID_PAYLOAD"
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		} else if (e instanceof SyntaxError) {
-			const response: ErrorResponse<string> = {
-				message: "Invalid JSON payload.",
-				errors: [e.message],
-				code: "INVALID_FORMAT"
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		} else {
-			const response: ErrorResponse<string> = {
-				message: "Unknown error.",
-				errors: [(e as Error).message],
-				code: "UNKNOWN_ERROR"
-			};
-			return c.json<ErrorResponse<string>>(response, 500);
-		}
-	}
-});

packages/backend/routes/user/index.ts

@@ -1,2 +0,0 @@
-export * from "./POST.ts";
-export * from "./session/[id]/GET.ts";

packages/backend/routes/user/session/[id]/GET.ts

@@ -1,32 +0,0 @@
-import { Context } from "hono";
-import { Bindings, BlankEnv } from "hono/types";
-import { ErrorResponse } from "src/schema";
-import { createHandlers } from "src/utils.ts";
-import { sqlCred } from "@core/db/dbNew";
-import { UserType } from "@core/db/schema";
-
-export const getUserByLoginSessionHandler = createHandlers(
-	async (c: Context<BlankEnv & { Bindings: Bindings }, "/user/session/:id">) => {
-		const id = c.req.param("id");
-		const users = await sqlCred<UserType[]>`
-    		SELECT u.*
-            FROM users u
-            JOIN login_sessions ls ON u.id = ls.uid
-            WHERE ls.id = ${id};
-        `;
-		if (users.length === 0) {
-			const response: ErrorResponse = {
-				message: "Cannot find user",
-				code: "ENTITY_NOT_FOUND",
-				errors: []
-			};
-			return c.json<ErrorResponse>(response, 404);
-		}
-		const user = users[0];
-		return c.json({
-			username: user.username,
-			nickname: user.nickname,
-			role: user.role
-		});
-	}
-);

packages/backend/routes/video/index.ts

@@ -1,2 +0,0 @@
-export * from "./[id]/info";
-export * from "./[id]/snapshots";

packages/backend/routes/videos/GET.ts

@@ -1,65 +0,0 @@
-import type { Context } from "hono";
-import { createHandlers } from "src/utils.ts";
-import type { BlankEnv, BlankInput } from "hono/types";
-import { number, object, ValidationError } from "yup";
-import { ErrorResponse } from "src/schema";
-import { startTime, endTime } from "hono/timing";
-import { getVideosInViewsRange } from "@/db/latestSnapshots";
-
-const SnapshotQueryParamsSchema = object({
-	min_views: number().integer().optional().positive(),
-	max_views: number().integer().optional().positive()
-});
-
-type ContextType = Context<BlankEnv, "/videos", BlankInput>;
-
-export const getVideosHanlder = createHandlers(async (c: ContextType) => {
-	startTime(c, "parse", "Parse the request");
-	try {
-		const queryParams = await SnapshotQueryParamsSchema.validate(c.req.query());
-		const { min_views, max_views } = queryParams;
-
-		if (!min_views && !max_views) {
-			const response: ErrorResponse<string> = {
-				code: "INVALID_QUERY_PARAMS",
-				message: "Invalid query parameters",
-				errors: ["Must provide one of these query parameters: min_views, max_views"]
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		}
-
-		endTime(c, "parse");
-
-		startTime(c, "db", "Query the database");
-
-		const minViews = min_views ? min_views : 0;
-		const maxViews = max_views ? max_views : 2147483647;
-
-		const result = await getVideosInViewsRange(minViews, maxViews);
-
-		endTime(c, "db");
-
-		const rows = result.map((row) => ({
-			...row,
-			aid: Number(row.aid)
-		}));
-
-		return c.json(rows);
-	} catch (e: unknown) {
-		if (e instanceof ValidationError) {
-			const response: ErrorResponse<string> = {
-				code: "INVALID_QUERY_PARAMS",
-				message: "Invalid query parameters",
-				errors: e.errors
-			};
-			return c.json<ErrorResponse<string>>(response, 400);
-		} else {
-			const response: ErrorResponse<unknown> = {
-				code: "UNKNOWN_ERROR",
-				message: "Unhandled error",
-				errors: [e]
-			};
-			return c.json<ErrorResponse<unknown>>(response, 500);
-		}
-	}
-});

packages/backend/routes/videos/index.ts

@@ -1 +0,0 @@
-export * from "./GET.ts";

packages/backend/src/main.ts

@@ -1,8 +1,8 @@
 import { Hono } from "hono";
 import type { TimingVariables } from "hono/timing";
 import { startServer } from "./startServer.ts";
-import { configureRoutes } from "./routing.ts";
+import { configureRoutes } from "routes";
-import { configureMiddleWares } from "./middleware.ts";
+import { configureMiddleWares } from "middleware";
 import { notFoundRoute } from "routes/404.ts";
 
 type Variables = TimingVariables;
@@ -15,4 +15,4 @@ configureRoutes(app);
 
 await startServer(app);
 
-export const VERSION = "0.6.0";
+export const VERSION = "0.4.6";

packages/backend/src/middleware.ts

@@ -1,24 +0,0 @@
-import { Hono } from "hono";
-import { timing } from "hono/timing";
-import { Variables } from "hono/types";
-import { pingHandler } from "routes/ping";
-import { logger } from "middleware/logger.ts";
-import { corsMiddleware } from "@/middleware/cors";
-import { contentType } from "middleware/contentType.ts";
-import { captchaMiddleware } from "middleware/captcha.ts";
-import { bodyLimitForPing } from "middleware/bodyLimits.ts";
-import { registerRateLimiter } from "middleware/rateLimiters.ts";
-import { preetifyResponse } from "middleware/preetifyResponse.ts";
-
-export function configureMiddleWares(app: Hono<{ Variables: Variables }>) {
-	app.use("*", corsMiddleware);
-
-	app.use("*", contentType);
-	app.use(timing());
-	app.use("*", preetifyResponse);
-	app.use("*", logger({}));
-
-	app.post("/user", registerRateLimiter);
-	app.post("/user", captchaMiddleware);
-	app.all("/ping", bodyLimitForPing, ...pingHandler);
-}

packages/backend/src/routing.ts

@@ -1,32 +0,0 @@
-import { rootHandler } from "routes";
-import { pingHandler } from "routes/ping";
-import { getUserByLoginSessionHandler, registerHandler } from "routes/user";
-import { videoInfoHandler, getSnapshotsHanlder } from "routes/video";
-import { Hono } from "hono";
-import { Variables } from "hono/types";
-import { createCaptchaSessionHandler, verifyChallengeHandler } from "routes/captcha";
-import { getCaptchaDifficultyHandler } from "routes/captcha/difficulty/GET.ts";
-import { getVideosHanlder } from "@/routes/videos";
-import { loginHandler } from "@/routes/login/session/POST";
-import { logoutHandler } from "@/routes/session";
-
-export function configureRoutes(app: Hono<{ Variables: Variables }>) {
-	app.get("/", ...rootHandler);
-	app.all("/ping", ...pingHandler);
-
-	app.get("/videos", ...getVideosHanlder);
-
-	app.get("/video/:id/snapshots", ...getSnapshotsHanlder);
-	app.get("/video/:id/info", ...videoInfoHandler);
-
-	app.post("/login/session", ...loginHandler);
-
-	app.delete("/session/:id", ...logoutHandler);
-
-	app.post("/user", ...registerHandler);
-	app.get("/user/session/:id", ...getUserByLoginSessionHandler);
-
-	app.post("/captcha/session", ...createCaptchaSessionHandler);
-	app.get("/captcha/:id/result", ...verifyChallengeHandler);
-	app.get("/captcha/difficulty", ...getCaptchaDifficultyHandler);
-}

packages/backend/src/schema.d.ts

@@ -1,66 +1,11 @@
-export type ErrorCode =
+type ErrorCode = "INVALID_QUERY_PARAMS" | "UNKNOWN_ERR" | "INVALID_PAYLOAD" | "INVALID_FORMAT" | "BODY_TOO_LARGE";
-	| "INVALID_QUERY_PARAMS"
-	| "UNKNOWN_ERROR"
-	| "INVALID_PAYLOAD"
-	| "INVALID_FORMAT"
-	| "INVALID_HEADER"
-	| "BODY_TOO_LARGE"
-	| "UNAUTHORIZED"
-	| "INVALID_CREDENTIALS"
-	| "ENTITY_NOT_FOUND"
-	| "SERVER_ERROR"
-	| "RATE_LIMIT_EXCEEDED"
-	| "ENTITY_EXISTS";
 
-export interface ErrorResponse<E = string> {
+export interface ErrorResponse<E> {
-	code: ErrorCode;
+	code: ErrorCode
 	message: string;
-	errors: E[] = [];
+	errors: E[];
-	i18n?: {
-		key: string;
-		values?: {
-			[key: string]: string | number | Date;
-		};
-	};
 }
 
 export interface StatusResponse {
 	message: string;
-}
+}
-
-export type CaptchaSessionResponse = ErrorResponse | CaptchaSessionRawResponse;
-
-interface CaptchaSessionRawResponse {
-	success: boolean;
-	id: string;
-	g: string;
-	n: string;
-	t: number;
-}
-
-export interface LoginResponse {
-    uid: number;
-    username: string;
-    nickname: string | null;
-    role: string;
-    token: string;
-}
-
-export interface SignUpResponse {
-    username: string;
-    token: string;
-}
-
-export interface UserResponse {
-    username: string;
-    nickname: string | null;
-    role: string;
-}
-
-export type CaptchaVerificationRawResponse = {
-	token: string;
-}
-
-export type CaptchaVerificationResponse =
-	| ErrorResponse
-	| CaptchaVerificationRawResponse;

packages/backend/src/startServer.ts

@@ -32,7 +32,7 @@ function logStartup(hostname: string, port: number, wasAutoIncremented: boolean,
 	console.log("\nPress Ctrl+C to quit.");
 }
 
-export async function startServer(app: Hono<{ Variables: Variables }>) {
+export async function startServer(app: Hono<{Variables: Variables }>) {
 	const NODE_ENV = process.env.NODE_ENV || "production";
 	const HOST = process.env.HOST ?? (NODE_ENV === "development" ? "0.0.0.0" : "127.0.0.1");
 	const PORT = process.env.PORT ? parseInt(process.env.PORT, 10) : undefined;

packages/backend/src/types.d.ts

@@ -1 +0,0 @@
-export * from "./schema";

packages/backend/tsconfig.json

@@ -10,7 +10,6 @@
 		"skipLibCheck": true,
 		"paths": {
 			"@core/*": ["../core/*"],
-			"@/*": ["./*"],
 			"@crawler/*": ["../crawler/*"]
 		},
 		"allowSyntheticDefaultImports": true,

packages/core/db/dbNew.ts

@@ -1,8 +1,6 @@
 import postgres from "postgres";
-import { postgresConfigCred, postgresConfig } from "./pgConfigNew";
+import { postgresConfigNpm } from "./pgConfigNew";
 
-export const sql = postgres(postgresConfig);
+export const sql = postgres(postgresConfigNpm);
 
-export const sqlCred = postgres(postgresConfigCred);
+export const sqlTest = postgres(postgresConfigNpm);
-
-export const sqlTest = postgres(postgresConfig);

packages/core/db/pgConfigNew.ts

@@ -1,23 +1,27 @@
 const requiredEnvVars = ["DB_HOST", "DB_NAME", "DB_USER", "DB_PASSWORD", "DB_PORT", "DB_NAME_CRED"];
 
-const getEnvVar = (key: string) => {
+const unsetVars = requiredEnvVars.filter((key) => process.env[key] === undefined);
-	return process.env[key] || import.meta.env[key];
-};
-
-const unsetVars = requiredEnvVars.filter((key) => getEnvVar(key) === undefined);
 
 if (unsetVars.length > 0) {
 	throw new Error(`Missing required environment variables: ${unsetVars.join(", ")}`);
 }
 
-const databaseHost = getEnvVar("DB_HOST")!;
+const databaseHost = process.env["DB_HOST"]!;
-const databaseName = getEnvVar("DB_NAME");
+const databaseName = process.env["DB_NAME"];
-const databaseNameCred = getEnvVar("DB_NAME_CRED")!;
+const databaseNameCred = process.env["DB_NAME_CRED"]!;
-const databaseUser = getEnvVar("DB_USER")!;
+const databaseUser = process.env["DB_USER"]!;
-const databasePassword = getEnvVar("DB_PASSWORD")!;
+const databasePassword = process.env["DB_PASSWORD"]!;
-const databasePort = getEnvVar("DB_PORT")!;
+const databasePort = process.env["DB_PORT"]!;
 
 export const postgresConfig = {
+	hostname: databaseHost,
+	port: parseInt(databasePort),
+	database: databaseName,
+	user: databaseUser,
+	password: databasePassword
+};
+
+export const postgresConfigNpm = {
 	host: databaseHost,
 	port: parseInt(databasePort),
 	database: databaseName,

packages/core/db/schema.d.ts

@@ -1,3 +1,16 @@
+export interface AllDataType {
+	id: number;
+	aid: number;
+	bvid: string | null;
+	description: string | null;
+	uid: number | null;
+	tags: string | null;
+	title: string | null;
+	published_at: string | null;
+	duration: number;
+	created_at: string | null;
+}
+
 export interface BiliUserType {
 	id: number;
 	uid: number;
@@ -8,7 +21,7 @@ export interface BiliUserType {
 
 export interface VideoSnapshotType {
 	id: number;
-	created_at: Date;
+	created_at: string;
 	views: number;
 	coins: number;
 	likes: number;
@@ -35,33 +48,8 @@ export interface SnapshotScheduleType {
 	id: number;
 	aid: number;
 	type?: string;
-	created_at: Date;
+	created_at: string;
-	started_at?: Date;
+	started_at?: string;
-	finished_at?: Date;
+	finished_at?: string;
 	status: string;
 }
-
-export interface UserType {
-	id: number;
-	username: string;
-	nickname: string | null;
-	password: string;
-	unq_id: string;
-	role: string;
-	created_at: Date;
-}
-
-export interface BiliVideoMetadataType {
-	id: number;
-	aid: number;
-	bvid: string | null;
-	description: string | null;
-	uid: number | null;
-	tags: string | null;
-	title: string | null;
-	published_at: Date | null;
-	duration: number | null;
-	created_at: Date;
-	status: number;
-	cover_url: string | null;
-}

packages/core/db/videoSnapshot.ts

@@ -0,0 +1,62 @@
+import type { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
+import type { VideoSnapshotType } from "./schema.d.ts";
+
+export async function getVideoSnapshots(
+	client: Client,
+	aid: number,
+	limit: number,
+	pageOrOffset: number,
+	reverse: boolean,
+	mode: "page" | "offset" = "page",
+) {
+	const offset = mode === "page" ? (pageOrOffset - 1) * limit : pageOrOffset;
+	const queryDesc: string = `
+        SELECT *
+        FROM video_snapshot
+        WHERE aid = $1
+        ORDER BY created_at DESC
+        LIMIT $2
+        OFFSET $3
+    `;
+	const queryAsc: string = `
+        SELECT *
+        FROM video_snapshot
+        WHERE aid = $1
+        ORDER BY created_at
+        LIMIT $2 OFFSET $3
+    `;
+	const query = reverse ? queryAsc : queryDesc;
+	const queryResult = await client.queryObject<VideoSnapshotType>(query, [aid, limit, offset]);
+	return queryResult.rows;
+}
+
+export async function getVideoSnapshotsByBV(
+	client: Client,
+	bv: string,
+	limit: number,
+	pageOrOffset: number,
+	reverse: boolean,
+	mode: "page" | "offset" = "page",
+) {
+	const offset = mode === "page" ? (pageOrOffset - 1) * limit : pageOrOffset;
+	const queryAsc = `
+        SELECT vs.*
+        FROM video_snapshot vs
+        JOIN bilibili_metadata bm ON vs.aid = bm.aid
+        WHERE bm.bvid = $1
+        ORDER BY vs.created_at
+        LIMIT $2
+        OFFSET $3
+    `;
+	const queryDesc: string = `
+        SELECT *
+        FROM video_snapshot vs
+        JOIN bilibili_metadata bm ON vs.aid = bm.aid
+        WHERE bm.bvid = $1
+        ORDER BY vs.created_at DESC
+        LIMIT $2 OFFSET $3
+    `;
+	const query = reverse ? queryAsc : queryDesc;
+	const queryResult = await client.queryObject<VideoSnapshotType>(query, [bv, limit, offset]);
+	return queryResult.rows;
+}

packages/core/deno.json

@@ -0,0 +1,17 @@
+{
+    "name": "@cvsa/core",
+    "exports": "./main.ts",
+    "imports": {
+        "ioredis": "npm:ioredis",
+        "log/": "./log/",
+        "db/": "./db/",
+        "$std/": "https://deno.land/std@0.216.0/",
+        "mq/": "./mq/",
+        "chalk": "npm:chalk",
+        "winston": "npm:winston",
+        "logform": "npm:logform",
+        "@core/": "./",
+        "child_process": "node:child_process",
+        "util": "node:util"
+    }
+}

packages/core/index.ts

@@ -1 +0,0 @@
-export * from "./db/dbNew";

packages/core/lib/randomID.ts

@@ -1,15 +0,0 @@
-export function generateRandomId(length: number): string {
-	const characters = 'abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789';
-	const charactersLength = characters.length;
-	const randomBytes = new Uint8Array(length);
-
-	crypto.getRandomValues(randomBytes);
-
-	let result = '';
-	for (let i = 0; i < length; i++) {
-		const randomIndex = randomBytes[i] % charactersLength;
-		result += characters.charAt(randomIndex);
-	}
-
-	return result;
-}

packages/core/main.ts

@@ -0,0 +1 @@
+export const DB_VERSION = 10;

packages/core/mq/multipleRateLimiter.ts

@@ -1,55 +0,0 @@
-import { RateLimiter as Limiter } from "@koshnic/ratelimit";
-import { redis } from "@core/db/redis.ts";
-
-export interface RateLimiterConfig {
-    duration: number;
-    max: number;
-}
-
-export class RateLimiterError extends Error {
-    public code: string;
-    constructor(message: string) {
-        super(message);
-        this.name = "RateLimiterError";
-        this.code = "RATE_LIMIT_EXCEEDED";
-    }
-}
-
-export class MultipleRateLimiter {
-    private readonly name: string;
-    private readonly configs: RateLimiterConfig[] = [];
-    private readonly limiter: Limiter;
-
-    /*
-     * @param name The name of the rate limiter
-     * @param configs The configuration of the rate limiter, containing:
-     * - duration: The duration of window in seconds
-     * - max: The maximum number of tokens allowed in the window
-     */
-    constructor(
-        name: string,
-        configs: RateLimiterConfig[]
-    ) {
-        this.configs = configs;
-        this.limiter = new Limiter(redis);
-        this.name = name;
-    }
-
-    /*
-     * Trigger an event in the rate limiter
-     */
-    async trigger(shouldThrow = true): Promise<void> {
-        for (let i = 0; i < this.configs.length; i++) {
-            const { duration, max } = this.configs[i];
-            const { allowed } = await this.limiter.allow(`cvsa:${this.name}_${i}`, {
-                burst: max,
-                ratePerPeriod: max,
-                period: duration,
-                cost: 1
-            });
-            if (!allowed && shouldThrow) {
-                throw new RateLimiterError("Rate limit exceeded")
-            }
-        }
-    }
-}

packages/core/mq/rateLimiter.ts

@@ -0,0 +1,56 @@
+import type { SlidingWindow } from "./slidingWindow.ts";
+
+export interface RateLimiterConfig {
+	window: SlidingWindow;
+	max: number;
+}
+
+export class RateLimiter {
+	private readonly configs: RateLimiterConfig[];
+	private readonly configEventNames: string[];
+
+	/*
+	 * @param name The name of the rate limiter
+	 * @param configs The configuration of the rate limiter, containing:
+	 * - window: The sliding window to use
+	 * - max: The maximum number of events allowed in the window
+	 */
+	constructor(name: string, configs: RateLimiterConfig[]) {
+		this.configs = configs;
+		this.configEventNames = configs.map((_, index) => `${name}_config_${index}`);
+	}
+
+	/*
+	 * Check if the event has reached the rate limit
+	 */
+	async getAvailability(): Promise<boolean> {
+		for (let i = 0; i < this.configs.length; i++) {
+			const config = this.configs[i];
+			const eventName = this.configEventNames[i];
+			const count = await config.window.count(eventName);
+			if (count >= config.max) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	/*
+	 * Trigger an event in the rate limiter
+	 */
+	async trigger(): Promise<void> {
+		for (let i = 0; i < this.configs.length; i++) {
+			const config = this.configs[i];
+			const eventName = this.configEventNames[i];
+			await config.window.event(eventName);
+		}
+	}
+
+	async clear(): Promise<void> {
+		for (let i = 0; i < this.configs.length; i++) {
+			const config = this.configs[i];
+			const eventName = this.configEventNames[i];
+			await config.window.clear(eventName);
+		}
+	}
+}

packages/core/mq/slidingWindow.ts

@@ -32,20 +32,14 @@ export class SlidingWindow {
 
 	/*
 	 * Count the number of events in the sliding window
-	 * @param {string} eventName The name of the event
+	 * @param eventName The name of the event
-	 * @param {number} [duration] The duration of the window in seconds
 	 */
-	async count(eventName: string, duration?: number): Promise<number> {
+	async count(eventName: string): Promise<number> {
 		const key = `cvsa:sliding_window:${eventName}`;
 		const now = Date.now();
 
 		// Remove timestamps outside the window
 		await this.redis.zremrangebyscore(key, 0, now - this.windowSize);
-
-		if (duration) {
-			return this.redis.zcount(key, now - duration * 1000, now);
-		}
-
 		// Get the number of timestamps in the window
 		return this.redis.zcard(key);
 	}
@@ -54,4 +48,4 @@ export class SlidingWindow {
 		const key = `cvsa:sliding_window:${eventName}`;
 		return this.redis.del(key);
 	}
-}
+}

packages/core/net/bilibili.d.ts

@@ -38,10 +38,6 @@ interface VideoInfoData {
 	ctime: number;
 	desc: string;
 	desc_v2: string;
-	tname: string;
-	tid: number;
-	tid_v2: number;
-	tname_v2: string;
 	state: number;
 	duration: number;
 	owner: {

packages/core/net/delegate.ts

@@ -1,7 +1,9 @@
 import logger from "@core/log/logger.ts";
-import { MultipleRateLimiter, RateLimiterError, type RateLimiterConfig } from "@core/mq/multipleRateLimiter.ts";
+import { RateLimiter, type RateLimiterConfig } from "mq/rateLimiter.ts";
+import { SlidingWindow } from "mq/slidingWindow.ts";
+import { redis } from "db/redis.ts";
 import { ReplyError } from "ioredis";
-import { SECOND } from "@core/const/time.ts";
+import { SECOND } from "../const/time.ts";
 import { spawn, SpawnOptions } from "child_process";
 
 export function spawnPromise(
@@ -71,11 +73,11 @@ export class NetSchedulerError extends Error {
 }
 
 type LimiterMap = {
-	[name: string]: MultipleRateLimiter;
+	[name: string]: RateLimiter;
 };
 
 type OptionalLimiterMap = {
-	[name: string]: MultipleRateLimiter | null;
+	[name: string]: RateLimiter | null;
 };
 
 type TaskMap = {
@@ -119,23 +121,20 @@ class NetworkDelegate {
 		const proxies = this.getTaskProxies(taskName);
 		for (const proxyName of proxies) {
 			const limiterId = "proxy-" + proxyName + "-" + taskName;
-			this.proxyLimiters[limiterId] = config ? new MultipleRateLimiter(limiterId, config) : null;
+			this.proxyLimiters[limiterId] = config ? new RateLimiter(limiterId, config) : null;
 		}
 	}
 
-	async triggerLimiter(task: string, proxy: string, force: boolean = false): Promise<void> {
+	async triggerLimiter(task: string, proxy: string): Promise<void> {
 		const limiterId = "proxy-" + proxy + "-" + task;
 		const providerLimiterId = "provider-" + proxy + "-" + this.tasks[task].provider;
 		try {
-			await this.proxyLimiters[limiterId]?.trigger(!force);
+			await this.proxyLimiters[limiterId]?.trigger();
-			await this.providerLimiters[providerLimiterId]?.trigger(!force);
+			await this.providerLimiters[providerLimiterId]?.trigger();
 		} catch (e) {
 			const error = e as Error;
 			if (e instanceof ReplyError) {
 				logger.error(error, "redis");
-			} else if (e instanceof RateLimiterError) {
-				// Re-throw it to ensure this.request can catch it
-				throw e;
 			}
 			logger.warn(`Unhandled error: ${error.message}`, "mq", "proxyRequest");
 		}
@@ -150,7 +149,7 @@ class NetworkDelegate {
 		}
 		for (const proxyName of bindProxies) {
 			const limiterId = "provider-" + proxyName + "-" + providerName;
-			this.providerLimiters[limiterId] = new MultipleRateLimiter(limiterId, config);
+			this.providerLimiters[limiterId] = new RateLimiter(limiterId, config);
 		}
 	}
 
@@ -169,15 +168,9 @@ class NetworkDelegate {
 		// find a available proxy
 		const proxiesNames = this.getTaskProxies(task);
 		for (const proxyName of shuffleArray(proxiesNames)) {
-			try {
+			if (await this.getProxyAvailability(proxyName, task)) {
 				return await this.proxyRequest<R>(url, proxyName, task, method);
 			}
-			catch (e) {
-				if (e instanceof RateLimiterError) {
-					continue;
-				}
-				throw e;
-			}
 		}
 		throw new NetSchedulerError("No proxy is available currently.", "NO_PROXY_AVAILABLE");
 	}
@@ -209,8 +202,16 @@ class NetworkDelegate {
 			throw new NetSchedulerError(`Proxy "${proxyName}" not found`, "PROXY_NOT_FOUND");
 		}
 
-		await this.triggerLimiter(task, proxyName, force);
+		if (!force) {
+			const isAvailable = await this.getProxyAvailability(proxyName, task);
+			const limiter = "proxy-" + proxyName + "-" + task;
+			if (!isAvailable) {
+				throw new NetSchedulerError(`Proxy "${limiter}" is rate limited`, "PROXY_RATE_LIMITED");
+			}
+		}
+
 		const result = await this.makeRequest<R>(url, proxy, method);
+		await this.triggerLimiter(task, proxyName);
 		return result;
 	}
 
@@ -225,6 +226,32 @@ class NetworkDelegate {
 		}
 	}
 
+	private async getProxyAvailability(proxyName: string, taskName: string): Promise<boolean> {
+		try {
+			const task = this.tasks[taskName];
+			const provider = task.provider;
+			const proxyLimiterId = "proxy-" + proxyName + "-" + task;
+			const providerLimiterId = "provider-" + proxyName + "-" + provider;
+			if (!this.proxyLimiters[proxyLimiterId]) {
+				const providerLimiter = this.providerLimiters[providerLimiterId];
+				return await providerLimiter.getAvailability();
+			}
+			const proxyLimiter = this.proxyLimiters[proxyLimiterId];
+			const providerLimiter = this.providerLimiters[providerLimiterId];
+			const providerAvailable = await providerLimiter.getAvailability();
+			const proxyAvailable = await proxyLimiter.getAvailability();
+			return providerAvailable && proxyAvailable;
+		} catch (e) {
+			const error = e as Error;
+			if (e instanceof ReplyError) {
+				logger.error(error, "redis");
+				return false;
+			}
+			logger.error(error, "mq", "getProxyAvailability");
+			return false;
+		}
+	}
+
 	private async nativeRequest<R>(url: string, method: string): Promise<R> {
 		try {
 			const controller = new AbortController();
@@ -289,37 +316,37 @@ class NetworkDelegate {
 const networkDelegate = new NetworkDelegate();
 const videoInfoRateLimiterConfig: RateLimiterConfig[] = [
 	{
-		duration: 0.3,
+		window: new SlidingWindow(redis, 0.3),
 		max: 1,
 	},
 	{
-		duration: 3,
+		window: new SlidingWindow(redis, 3),
 		max: 5,
 	},
 	{
-		duration: 30,
+		window: new SlidingWindow(redis, 30),
 		max: 30,
 	},
 	{
-		duration: 2 * 60,
+		window: new SlidingWindow(redis, 2 * 60),
 		max: 50,
 	},
 ];
 const biliLimiterConfig: RateLimiterConfig[] = [
 	{
-		duration: 1,
+		window: new SlidingWindow(redis, 1),
 		max: 6,
 	},
 	{
-		duration: 5,
+		window: new SlidingWindow(redis, 5),
 		max: 20,
 	},
 	{
-		duration: 30,
+		window: new SlidingWindow(redis, 30),
 		max: 100,
 	},
 	{
-		duration: 5 * 60,
+		window: new SlidingWindow(redis, 5 * 60),
 		max: 200,
 	},
 ];

packages/core/package.json

@@ -1,22 +1,13 @@
 {
-	"name": "@cvsa/core",
+  "name": "core",
-	"private": false,
+  "dependencies": {
-	"version": "0.0.10",
+    "chalk": "^5.4.1",
-	"scripts": {
+    "ioredis": "^5.6.1",
-		"test": "bun --env-file=.env.test run vitest",
+    "logform": "^2.7.0",
-		"build": "bun build ./index.ts --target node --outdir ./dist"
+    "postgres": "^3.4.5",
-	},
+    "winston": "^3.17.0"
-	"dependencies": {
+  },
-		"@koshnic/ratelimit": "^1.0.3",
+  "devDependencies": {
-		"chalk": "^5.4.1",
+    "@types/ioredis": "^5.0.0"
-		"ioredis": "^5.6.1",
+  }
-		"logform": "^2.7.0",
+}
-		"postgres": "^3.4.5",
-		"winston": "^3.17.0"
-	},
-	"devDependencies": {
-		"@types/ioredis": "^5.0.0"
-	},
-	"main": "./dist/index.js",
-	"types": "./types.d.ts"
-}

packages/core/test/lib/randomID.test.ts

@@ -1,18 +0,0 @@
-import { describe, expect, it } from "vitest";
-import { generateRandomId } from "@core/lib/randomID.ts";
-
-describe("generateRandomId", () => {
-	it("should generate an ID of the specified length", () => {
-		const length = 15;
-		const id = generateRandomId(length);
-		expect(id).toHaveLength(length);
-	});
-
-	it("should generate an ID containing only allowed characters", () => {
-		const allowedChars = "abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789";
-		const id = generateRandomId(20);
-		for (const char of id) {
-			expect(allowedChars).toContain(char);
-		}
-	});
-});

packages/core/types.d.ts

@@ -1,3 +0,0 @@
-export * from "./db/schema";
-export * from "./index";
-export * from "./net/bilibili";

packages/crawler/db/bilibili_metadata.ts

@@ -1,77 +1,77 @@
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
-import { BiliVideoMetadataType, BiliUserType } from "@core/db/schema";
+import { AllDataType, BiliUserType } from "@core/db/schema";
 import { AkariModelVersion } from "ml/const";
 
 export async function videoExistsInAllData(sql: Psql, aid: number) {
-	const rows = await sql<{ exists: boolean }[]>`
+    const rows = await sql<{ exists: boolean }[]>`
         SELECT EXISTS(SELECT 1 FROM bilibili_metadata WHERE aid = ${aid})
     `;
-	return rows[0].exists;
+    return rows[0].exists;
 }
 
 export async function userExistsInBiliUsers(sql: Psql, uid: number) {
-	const rows = await sql<{ exists: boolean }[]>`
+    const rows = await sql<{ exists: boolean }[]>`
         SELECT EXISTS(SELECT 1 FROM bilibili_user WHERE uid = ${uid})
     `;
-	return rows[0].exists;
+    return rows[0].exists;
 }
 
 export async function getUnlabelledVideos(sql: Psql) {
-	const rows = await sql<{ aid: number }[]>`
+    const rows = await sql<{ aid: number }[]>`
         SELECT a.aid FROM bilibili_metadata a LEFT JOIN labelling_result l ON a.aid = l.aid WHERE l.aid IS NULL
     `;
-	return rows.map((row) => row.aid);
+    return rows.map((row) => row.aid);
 }
 
 export async function insertVideoLabel(sql: Psql, aid: number, label: number) {
-	await sql`
+    await sql`
         INSERT INTO labelling_result (aid, label, model_version) VALUES (${aid}, ${label}, ${AkariModelVersion}) ON CONFLICT (aid, model_version) DO NOTHING
     `;
 }
 
 export async function getVideoInfoFromAllData(sql: Psql, aid: number) {
-	const rows = await sql<AllDataType[]>`
+    const rows = await sql<AllDataType[]>`
         SELECT * FROM bilibili_metadata WHERE aid = ${aid}
-  `;
+    `;
-	const row = rows[0];
+    const row = rows[0];
-	let authorInfo = "";
+    let authorInfo = "";
-	if (row.uid && (await userExistsInBiliUsers(sql, row.uid))) {
+    if (row.uid && await userExistsInBiliUsers(sql, row.uid)) {
-		const userRows = await sql<BiliUserType[]>`
+        const userRows = await sql<BiliUserType[]>`
             SELECT * FROM bilibili_user WHERE uid = ${row.uid}
         `;
-		const userRow = userRows[0];
+        const userRow = userRows[0];
-		if (userRow) {
+        if (userRow) {
-			authorInfo = userRow.desc;
+            authorInfo = userRow.desc;
-		}
+        }
-	}
+    }
-	return {
+    return {
-		title: row.title,
+        title: row.title,
-		description: row.description,
+        description: row.description,
-		tags: row.tags,
+        tags: row.tags,
-		author_info: authorInfo
+        author_info: authorInfo,
-	};
+    };
 }
 
 export async function getUnArchivedBiliUsers(sql: Psql) {
-	const rows = await sql<{ uid: number }[]>`
+    const rows = await sql<{ uid: number }[]>`
         SELECT ad.uid
         FROM bilibili_metadata ad
         LEFT JOIN bilibili_user bu ON ad.uid = bu.uid
         WHERE bu.uid IS NULL;
     `;
-	return rows.map((row) => row.uid);
+    return rows.map((row) => row.uid);
 }
 
 export async function setBiliVideoStatus(sql: Psql, aid: number, status: number) {
-	await sql`
+    await sql`
         UPDATE bilibili_metadata SET status = ${status} WHERE aid = ${aid}
     `;
 }
 
 export async function getBiliVideoStatus(sql: Psql, aid: number) {
-	const rows = await sql<{ status: number }[]>`
+    const rows = await sql<{ status: number }[]>`
         SELECT status FROM bilibili_metadata WHERE aid = ${aid}
     `;
-	if (rows.length === 0) return 0;
+    if (rows.length === 0) return 0;
-	return rows[0].status;
+    return rows[0].status;
-}
+}

packages/crawler/db/snapshot.ts

@@ -1,28 +1,28 @@
 import { LatestSnapshotType } from "@core/db/schema";
 import { SnapshotNumber } from "mq/task/getVideoStats.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 export async function getVideosNearMilestone(sql: Psql) {
 	const queryResult = await sql<LatestSnapshotType[]>`
-    	SELECT ls.*
+        SELECT ls.*
         FROM latest_video_snapshot ls
-                RIGHT JOIN songs ON songs.aid = ls.aid
+                 RIGHT JOIN songs ON songs.aid = ls.aid
         WHERE
             (views >= 50000 AND views < 100000) OR
             (views >= 900000 AND views < 1000000) OR
-            (views >= CEIL(views::float/1000000::float)*1000000-100000 AND views < CEIL(views::float/1000000::float)*1000000)
+            (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 >= CEIL(views::float/1000000::float)*1000000-100000 AND views < CEIL(views::float/1000000::float)*1000000)
+            (views >= 9900000 AND views < 10000000)
     `;
 	return queryResult.map((row) => {
 		return {
 			...row,
-			aid: Number(row.aid)
+			aid: Number(row.aid),
 		};
 	});
 }
@@ -40,7 +40,7 @@ export async function getLatestVideoSnapshot(sql: Psql, aid: number): Promise<nu
 		return {
 			...row,
 			aid: Number(row.aid),
-			time: new Date(row.time).getTime()
+			time: new Date(row.time).getTime(),
 		};
 	})[0];
 }

packages/crawler/db/snapshotSchedule.ts

@@ -4,7 +4,7 @@ import { MINUTE } from "@core/const/time.ts";
 import { redis } from "@core/db/redis.ts";
 import { Redis } from "ioredis";
 import { parseTimestampFromPsql } from "../utils/formatTimestampToPostgre.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 const REDIS_KEY = "cvsa:snapshot_window_counts";
 
@@ -19,8 +19,8 @@ export async function refreshSnapshotWindowCounts(sql: Psql, redisClient: Redis)
 	const startTime = now.getTime();
 
 	const result = await sql<{ window_start: Date; count: number }[]>`
-		SELECT
+		SELECT 
-		date_trunc('hour', started_at) +
+		date_trunc('hour', started_at) + 
 		(EXTRACT(minute FROM started_at)::int / 5 * INTERVAL '5 minutes') AS window_start,
 		COUNT(*) AS count
 		FROM snapshot_schedule
@@ -56,18 +56,30 @@ async function getWindowCount(redisClient: Redis, offset: number): Promise<numbe
 
 export async function snapshotScheduleExists(sql: Psql, id: number) {
 	const rows = await sql<{ id: number }[]>`
-		SELECT id
+		SELECT id 
-		FROM snapshot_schedule
+		FROM snapshot_schedule 
 		WHERE id = ${id}
 	`;
 	return rows.length > 0;
 }
 
+export async function videoHasActiveSchedule(sql: Psql, aid: number) {
+	const rows = await sql<{ status: string }[]>`
+		SELECT status
+		FROM snapshot_schedule 
+		WHERE aid = ${aid}
+			AND (status = 'pending' 
+				OR status = 'processing'
+			)
+	`
+	return rows.length > 0;
+}
+
 export async function videoHasActiveScheduleWithType(sql: Psql, aid: number, type: string) {
 	const rows = await sql<{ status: string }[]>`
-		SELECT status FROM snapshot_schedule
+		SELECT status FROM snapshot_schedule 
 		WHERE aid = ${aid}
-			AND (status = 'pending' OR status = 'processing')
+			AND (status = 'pending' OR status = 'processing') 
 			AND type = ${type}
 	`;
 	return rows.length > 0;
@@ -76,10 +88,10 @@ export async function videoHasActiveScheduleWithType(sql: Psql, aid: number, typ
 export async function videoHasProcessingSchedule(sql: Psql, aid: number) {
 	const rows = await sql<{ status: string }[]>`
 		SELECT status
-		FROM snapshot_schedule
+		FROM snapshot_schedule 
 		WHERE aid = ${aid}
 			AND status = 'processing'
-	`;
+	`
 	return rows.length > 0;
 }
 
@@ -88,9 +100,9 @@ export async function bulkGetVideosWithoutProcessingSchedules(sql: Psql, aids: n
 		SELECT aid
 		FROM snapshot_schedule
 		WHERE aid = ANY(${aids})
-			AND status != 'processing'
+			AND status != 'processing' 
 		GROUP BY aid
-	`;
+	`
 	return rows.map((row) => Number(row.aid));
 }
 
@@ -134,8 +146,8 @@ export async function findSnapshotBefore(sql: Psql, aid: number, targetTime: Dat
 
 export async function hasAtLeast2Snapshots(sql: Psql, aid: number) {
 	const res = await sql<{ count: number }[]>`
-		SELECT COUNT(*)
+		SELECT COUNT(*) 
-		FROM video_snapshot
+		FROM video_snapshot 
 		WHERE aid = ${aid}
 	`;
 	return res[0].count >= 2;
@@ -143,10 +155,10 @@ export async function hasAtLeast2Snapshots(sql: Psql, aid: number) {
 
 export async function getLatestSnapshot(sql: Psql, aid: number): Promise<Snapshot | null> {
 	const res = await sql<{ created_at: string; views: number }[]>`
-		SELECT created_at, views
+		SELECT created_at, views 
-		FROM video_snapshot
+		FROM video_snapshot 
 		WHERE aid = ${aid}
-		ORDER BY created_at DESC
+		ORDER BY created_at DESC 
 		LIMIT 1
 	`;
 	if (res.length === 0) return null;
@@ -182,8 +194,7 @@ export async function scheduleSnapshot(
 	aid: number,
 	type: string,
 	targetTime: number,
-	force: boolean = false,
+	force: boolean = false
-	adjustTime: boolean = true
 ) {
 	let adjustedTime = new Date(targetTime);
 	const hashActiveSchedule = await videoHasActiveScheduleWithType(sql, aid, type);
@@ -205,16 +216,16 @@ export async function scheduleSnapshot(
 		}
 	}
 	if (hashActiveSchedule && !force) return;
-	if (type !== "milestone" && type !== "new" && adjustTime) {
+	if (type !== "milestone" && type !== "new") {
 		adjustedTime = await adjustSnapshotTime(new Date(targetTime), 2000, redis);
 	}
 	logger.log(`Scheduled snapshot for ${aid} at ${adjustedTime.toISOString()}`, "mq", "fn:scheduleSnapshot");
 	return sql`
-		INSERT INTO snapshot_schedule
+		INSERT INTO snapshot_schedule 
-			(aid, type, started_at)
+			(aid, type, started_at) 
 			VALUES (
-				${aid},
+				${aid}, 
-				${type},
+				${type}, 
 				${adjustedTime.toISOString()}
 			)
 	`;
@@ -225,11 +236,10 @@ export async function bulkScheduleSnapshot(
 	aids: number[],
 	type: string,
 	targetTime: number,
-	force: boolean = false,
+	force: boolean = false
-	adjustTime: boolean = true
 ) {
 	for (const aid of aids) {
-		await scheduleSnapshot(sql, aid, type, targetTime, force, adjustTime);
+		await scheduleSnapshot(sql, aid, type, targetTime, force);
 	}
 }
 
@@ -282,23 +292,23 @@ export async function adjustSnapshotTime(
 }
 
 export async function getSnapshotsInNextSecond(sql: Psql) {
-	return sql<SnapshotScheduleType[]>`
+	const rows = await sql<SnapshotScheduleType[]>`
-        SELECT *
+		SELECT *
-        FROM snapshot_schedule
+		FROM snapshot_schedule
-        WHERE started_at <= NOW() + INTERVAL '1 seconds'
+		WHERE started_at <= NOW() + INTERVAL '1 seconds' AND status = 'pending' AND type != 'normal'
-          AND status = 'pending'
+		ORDER BY
-          AND type != 'normal'
+			CASE
-        ORDER BY CASE
+				WHEN type = 'milestone' THEN 0
-                     WHEN type = 'milestone' THEN 0
+				ELSE 1
-                     ELSE 1
+			END,
-                     END,
+			started_at
-                 started_at
+		LIMIT 10;
-        LIMIT 10;
+	`
-	`;
+	return rows;
 }
 
 export async function getBulkSnapshotsInNextSecond(sql: Psql) {
-	return sql<SnapshotScheduleType[]>`
+	const rows = await sql<SnapshotScheduleType[]>`
         SELECT *
         FROM snapshot_schedule
         WHERE (started_at <= NOW() + INTERVAL '15 seconds')
@@ -310,34 +320,38 @@ export async function getBulkSnapshotsInNextSecond(sql: Psql) {
                      END,
                  started_at
         LIMIT 1000;
-	`;
+	`
+	return rows;
 }
 
 export async function setSnapshotStatus(sql: Psql, id: number, status: string) {
-	return sql`
+	return await sql`
-        UPDATE snapshot_schedule
+		UPDATE snapshot_schedule SET status = ${status} WHERE id = ${id}
-        SET status = ${status}
-        WHERE id = ${id}
 	`;
 }
 
 export async function bulkSetSnapshotStatus(sql: Psql, ids: number[], status: string) {
-	return sql`
+	return await sql`
-        UPDATE snapshot_schedule
+		UPDATE snapshot_schedule SET status = ${status} WHERE id = ANY(${ids})
-        SET status = ${status}
-        WHERE id = ANY (${ids})
 	`;
 }
 
-export async function getVideosWithoutActiveSnapshotScheduleByType(sql: Psql, type: string) {
+export async function getVideosWithoutActiveSnapshotSchedule(sql: Psql) {
 	const rows = await sql<{ aid: string }[]>`
 		SELECT s.aid
 		FROM songs s
-		LEFT JOIN snapshot_schedule ss ON
+		LEFT JOIN snapshot_schedule ss ON s.aid = ss.aid AND (ss.status = 'pending' OR ss.status = 'processing')
-		    s.aid = ss.aid AND
-		    (ss.status = 'pending' OR ss.status = 'processing') AND
-		    ss.type = ${type}
 		WHERE ss.aid IS NULL
 	`;
 	return rows.map((r) => Number(r.aid));
 }
+
+export async function getAllVideosWithoutActiveSnapshotSchedule(psql: Psql) {
+	const rows = await psql<{ aid: number }[]>`
+		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
+	`
+	return rows.map((r) => Number(r.aid));
+}

packages/crawler/db/songs.ts

@@ -1,8 +1,8 @@
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 import { parseTimestampFromPsql } from "utils/formatTimestampToPostgre.ts";
 
 export async function getNotCollectedSongs(sql: Psql) {
-	const rows = await sql<{ aid: number }[]>`
+    const rows = await sql<{ aid: number }[]>`
         SELECT lr.aid
         FROM labelling_result lr
         WHERE lr.label != 0
@@ -12,28 +12,28 @@ export async function getNotCollectedSongs(sql: Psql) {
             WHERE s.aid = lr.aid
         );
     `;
-	return rows.map((row) => row.aid);
+    return rows.map((row) => row.aid);
 }
 
 export async function aidExistsInSongs(sql: Psql, aid: number) {
-	const rows = await sql<{ exists: boolean }[]>`
+    const rows = await sql<{ exists: boolean }[]>`
         SELECT EXISTS (
             SELECT 1
             FROM songs
             WHERE aid = ${aid}
         );
     `;
-	return rows[0].exists;
+    return rows[0].exists;
 }
 
 export async function getSongsPublihsedAt(sql: Psql, aid: number) {
-	const rows = await sql<{ published_at: string }[]>`
+    const rows = await sql<{ published_at: string }[]>`
         SELECT published_at
         FROM songs
         WHERE aid = ${aid};
     `;
-	if (rows.length === 0) {
+    if (rows.length === 0) {
-		return null;
+        return null;
-	}
+    }
-	return parseTimestampFromPsql(rows[0].published_at);
+    return parseTimestampFromPsql(rows[0].published_at);
-}
+}

packages/crawler/global.d.ts

@@ -1,3 +1,3 @@
 import type postgres from "postgres";
 
-export type Psql = postgres.Sql;
+export type Psql = postgres.Sql<{}>;

packages/crawler/ml/akari.ts

@@ -16,8 +16,8 @@ class AkariProto extends AIManager {
 	constructor() {
 		super();
 		this.models = {
-			classifier: onnxClassifierPath,
+			"classifier": onnxClassifierPath,
-			embedding: onnxEmbeddingPath
+			"embedding": onnxEmbeddingPath,
 		};
 	}
 
@@ -55,7 +55,7 @@ class AkariProto extends AIManager {
 
 		const { input_ids } = await tokenizer(texts, {
 			add_special_tokens: false,
-			return_tensor: false
+			return_tensor: false,
 		});
 
 		const cumsum = (arr: number[]): number[] =>
@@ -66,9 +66,9 @@ class AkariProto extends AIManager {
 
 		const inputs = {
 			input_ids: new ort.Tensor("int64", new BigInt64Array(flattened_input_ids.map(BigInt)), [
-				flattened_input_ids.length
+				flattened_input_ids.length,
 			]),
-			offsets: new ort.Tensor("int64", new BigInt64Array(offsets.map(BigInt)), [offsets.length])
+			offsets: new ort.Tensor("int64", new BigInt64Array(offsets.map(BigInt)), [offsets.length]),
 		};
 
 		const { embeddings } = await session.run(inputs);
@@ -77,14 +77,21 @@ class AkariProto extends AIManager {
 
 	private async runClassification(embeddings: number[]): Promise<number[]> {
 		const session = this.getModelSession("classifier");
-		const inputTensor = new ort.Tensor(Float32Array.from(embeddings), [1, 3, 1024]);
+		const inputTensor = new ort.Tensor(
+			Float32Array.from(embeddings),
+			[1, 3, 1024],
+		);
 
 		const { logits } = await session.run({ channel_features: inputTensor });
 		return this.softmax(logits.data as Float32Array);
 	}
 
 	public async classifyVideo(title: string, description: string, tags: string, aid?: number): Promise<number> {
-		const embeddings = await this.getJinaEmbeddings1024([title, description, tags]);
+		const embeddings = await this.getJinaEmbeddings1024([
+			title,
+			description,
+			tags,
+		]);
 		const probabilities = await this.runClassification(embeddings);
 		if (aid) {
 			logger.log(`Prediction result for aid: ${aid}: [${probabilities.map((p) => p.toFixed(5))}]`, "ml");

packages/crawler/ml/benchmark.ts

@@ -0,0 +1,179 @@
+import { AutoTokenizer, PreTrainedTokenizer } from "@huggingface/transformers";
+import * as ort from "onnxruntime";
+
+function softmax(logits: Float32Array): number[] {
+	const maxLogit = Math.max(...logits);
+	const exponents = logits.map((logit) => Math.exp(logit - maxLogit));
+	const sumOfExponents = exponents.reduce((sum, exp) => sum + exp, 0);
+	return Array.from(exponents.map((exp) => exp / sumOfExponents));
+}
+
+// 配置参数
+const sentenceTransformerModelName = "alikia2x/jina-embedding-v3-m2v-1024";
+const onnxClassifierPath = "./model/video_classifier_v3_17.onnx";
+const onnxEmbeddingPath = "./model/embedding_original.onnx";
+const testDataPath = "./data/filter/test1.jsonl";
+
+// 初始化会话
+const [sessionClassifier, sessionEmbedding] = await Promise.all([
+	ort.InferenceSession.create(onnxClassifierPath),
+	ort.InferenceSession.create(onnxEmbeddingPath),
+]);
+
+let tokenizer: PreTrainedTokenizer;
+
+// 初始化分词器
+async function loadTokenizer() {
+	const tokenizerConfig = { local_files_only: true };
+	tokenizer = await AutoTokenizer.from_pretrained(sentenceTransformerModelName, tokenizerConfig);
+}
+
+// 新的嵌入生成函数（使用ONNX）
+async function getONNXEmbeddings(texts: string[], session: ort.InferenceSession): Promise<number[]> {
+	const { input_ids } = await tokenizer(texts, {
+		add_special_tokens: false,
+		return_tensor: false,
+	});
+
+	// 构造输入参数
+	const cumsum = (arr: number[]): number[] =>
+		arr.reduce((acc: number[], num: number, i: number) => [...acc, num + (acc[i - 1] || 0)], []);
+
+	const offsets: number[] = [0, ...cumsum(input_ids.slice(0, -1).map((x: string) => x.length))];
+	const flattened_input_ids = input_ids.flat();
+
+	// 准备ONNX输入
+	const inputs = {
+		input_ids: new ort.Tensor("int64", new BigInt64Array(flattened_input_ids.map(BigInt)), [
+			flattened_input_ids.length,
+		]),
+		offsets: new ort.Tensor("int64", new BigInt64Array(offsets.map(BigInt)), [offsets.length]),
+	};
+
+	// 执行推理
+	const { embeddings } = await session.run(inputs);
+	return Array.from(embeddings.data as Float32Array);
+}
+
+// 分类推理函数
+async function runClassification(embeddings: number[]): Promise<number[]> {
+	const inputTensor = new ort.Tensor(
+		Float32Array.from(embeddings),
+		[1, 3, 1024],
+	);
+
+	const { logits } = await sessionClassifier.run({ channel_features: inputTensor });
+	return softmax(logits.data as Float32Array);
+}
+
+// 指标计算函数
+function calculateMetrics(labels: number[], predictions: number[], elapsedTime: number): {
+	accuracy: number;
+	precision: number;
+	recall: number;
+	f1: number;
+	"Class 0 Prec": number;
+	speed: string;
+} {
+	// 输出label和prediction不一样的index列表
+	const arr = [];
+	for (let i = 0; i < labels.length; i++) {
+		if (labels[i] !== predictions[i] && predictions[i] == 0) {
+			arr.push([i + 1, labels[i], predictions[i]]);
+		}
+	}
+	console.log(arr);
+	// 初始化混淆矩阵
+	const classCount = Math.max(...labels, ...predictions) + 1;
+	const matrix = Array.from({ length: classCount }, () => Array.from({ length: classCount }, () => 0));
+
+	// 填充矩阵
+	labels.forEach((trueLabel, i) => {
+		matrix[trueLabel][predictions[i]]++;
+	});
+
+	// 计算各指标
+	let totalTP = 0, totalFP = 0, totalFN = 0;
+
+	for (let c = 0; c < classCount; c++) {
+		const TP = matrix[c][c];
+		const FP = matrix.flatMap((row, i) => i === c ? [] : [row[c]]).reduce((a, b) => a + b, 0);
+		const FN = matrix[c].filter((_, i) => i !== c).reduce((a, b) => a + b, 0);
+
+		totalTP += TP;
+		totalFP += FP;
+		totalFN += FN;
+	}
+
+	const precision = totalTP / (totalTP + totalFP);
+	const recall = totalTP / (totalTP + totalFN);
+	const f1 = 2 * (precision * recall) / (precision + recall) || 0;
+
+	// 计算Class 0 Precision
+	const class0TP = matrix[0][0];
+	const class0FP = matrix.flatMap((row, i) => i === 0 ? [] : [row[0]]).reduce((a, b) => a + b, 0);
+	const class0Precision = class0TP / (class0TP + class0FP) || 0;
+
+	return {
+		accuracy: labels.filter((l, i) => l === predictions[i]).length / labels.length,
+		precision,
+		recall,
+		f1,
+		speed: `${(labels.length / (elapsedTime / 1000)).toFixed(1)} samples/sec`,
+		"Class 0 Prec": class0Precision,
+	};
+}
+
+// 改造后的评估函数
+async function evaluateModel(session: ort.InferenceSession): Promise<{
+	accuracy: number;
+	precision: number;
+	recall: number;
+	f1: number;
+	"Class 0 Prec": number;
+}> {
+	const data = await Deno.readTextFile(testDataPath);
+	const samples = data.split("\n")
+		.map((line) => {
+			try {
+				return JSON.parse(line);
+			} catch {
+				return null;
+			}
+		})
+		.filter(Boolean);
+
+	const allPredictions: number[] = [];
+	const allLabels: number[] = [];
+
+	const t = new Date().getTime();
+	for (const sample of samples) {
+		try {
+			const embeddings = await getONNXEmbeddings([
+				sample.title,
+				sample.description,
+				sample.tags.join(","),
+			], session);
+
+			const probabilities = await runClassification(embeddings);
+			allPredictions.push(probabilities.indexOf(Math.max(...probabilities)));
+			allLabels.push(sample.label);
+		} catch (error) {
+			console.error("Processing error:", error);
+		}
+	}
+	const elapsed = new Date().getTime() - t;
+
+	return calculateMetrics(allLabels, allPredictions, elapsed);
+}
+
+// 主函数
+async function main() {
+	await loadTokenizer();
+
+	const metrics = await evaluateModel(sessionEmbedding);
+	console.log("Model Metrics:");
+	console.table(metrics);
+}
+
+await main();

packages/crawler/ml/const.ts

@@ -1 +1 @@
-export const AkariModelVersion = "3.17";
+export const AkariModelVersion = "3.17";

packages/crawler/ml/manager.ts

@@ -6,7 +6,8 @@ export class AIManager {
 	public sessions: { [key: string]: ort.InferenceSession } = {};
 	public models: { [key: string]: string } = {};
 
-	constructor() {}
+	constructor() {
+	}
 
 	public async init() {
 		const modelKeys = Object.keys(this.models);

packages/crawler/ml/quant_benchmark.ts

@@ -0,0 +1,171 @@
+import { AutoTokenizer, PreTrainedTokenizer } from "@huggingface/transformers";
+import * as ort from "onnxruntime";
+
+function softmax(logits: Float32Array): number[] {
+	const maxLogit = Math.max(...logits);
+	const exponents = logits.map((logit) => Math.exp(logit - maxLogit));
+	const sumOfExponents = exponents.reduce((sum, exp) => sum + exp, 0);
+	return Array.from(exponents.map((exp) => exp / sumOfExponents));
+}
+
+// 配置参数
+const sentenceTransformerModelName = "alikia2x/jina-embedding-v3-m2v-1024";
+const onnxClassifierPath = "./model/video_classifier_v3_11.onnx";
+const onnxEmbeddingOriginalPath = "./model/embedding_original.onnx";
+const onnxEmbeddingQuantizedPath = "./model/embedding_original.onnx";
+
+// 初始化会话
+const [sessionClassifier, sessionEmbeddingOriginal, sessionEmbeddingQuantized] = await Promise.all([
+	ort.InferenceSession.create(onnxClassifierPath),
+	ort.InferenceSession.create(onnxEmbeddingOriginalPath),
+	ort.InferenceSession.create(onnxEmbeddingQuantizedPath),
+]);
+
+let tokenizer: PreTrainedTokenizer;
+
+// 初始化分词器
+async function loadTokenizer() {
+	const tokenizerConfig = { local_files_only: true };
+	tokenizer = await AutoTokenizer.from_pretrained(sentenceTransformerModelName, tokenizerConfig);
+}
+
+// 新的嵌入生成函数（使用ONNX）
+async function getONNXEmbeddings(texts: string[], session: ort.InferenceSession): Promise<number[]> {
+	const { input_ids } = await tokenizer(texts, {
+		add_special_tokens: false,
+		return_tensor: false,
+	});
+
+	// 构造输入参数
+	const cumsum = (arr: number[]): number[] =>
+		arr.reduce((acc: number[], num: number, i: number) => [...acc, num + (acc[i - 1] || 0)], []);
+
+	const offsets: number[] = [0, ...cumsum(input_ids.slice(0, -1).map((x: string) => x.length))];
+	const flattened_input_ids = input_ids.flat();
+
+	// 准备ONNX输入
+	const inputs = {
+		input_ids: new ort.Tensor("int64", new BigInt64Array(flattened_input_ids.map(BigInt)), [
+			flattened_input_ids.length,
+		]),
+		offsets: new ort.Tensor("int64", new BigInt64Array(offsets.map(BigInt)), [offsets.length]),
+	};
+
+	// 执行推理
+	const { embeddings } = await session.run(inputs);
+	return Array.from(embeddings.data as Float32Array);
+}
+
+// 分类推理函数
+async function runClassification(embeddings: number[]): Promise<number[]> {
+	const inputTensor = new ort.Tensor(
+		Float32Array.from(embeddings),
+		[1, 4, 1024],
+	);
+
+	const { logits } = await sessionClassifier.run({ channel_features: inputTensor });
+	return softmax(logits.data as Float32Array);
+}
+
+// 指标计算函数
+function calculateMetrics(labels: number[], predictions: number[], elapsedTime: number): {
+	accuracy: number;
+	precision: number;
+	recall: number;
+	f1: number;
+	speed: string;
+} {
+	// 初始化混淆矩阵
+	const classCount = Math.max(...labels, ...predictions) + 1;
+	const matrix = Array.from({ length: classCount }, () => Array.from({ length: classCount }, () => 0));
+
+	// 填充矩阵
+	labels.forEach((trueLabel, i) => {
+		matrix[trueLabel][predictions[i]]++;
+	});
+
+	// 计算各指标
+	let totalTP = 0, totalFP = 0, totalFN = 0;
+
+	for (let c = 0; c < classCount; c++) {
+		const TP = matrix[c][c];
+		const FP = matrix.flatMap((row, i) => i === c ? [] : [row[c]]).reduce((a, b) => a + b, 0);
+		const FN = matrix[c].filter((_, i) => i !== c).reduce((a, b) => a + b, 0);
+
+		totalTP += TP;
+		totalFP += FP;
+		totalFN += FN;
+	}
+
+	const precision = totalTP / (totalTP + totalFP);
+	const recall = totalTP / (totalTP + totalFN);
+	const f1 = 2 * (precision * recall) / (precision + recall) || 0;
+
+	return {
+		accuracy: labels.filter((l, i) => l === predictions[i]).length / labels.length,
+		precision,
+		recall,
+		f1,
+		speed: `${(labels.length / (elapsedTime / 1000)).toFixed(1)} samples/sec`,
+	};
+}
+
+// 改造后的评估函数
+async function evaluateModel(session: ort.InferenceSession): Promise<{
+	accuracy: number;
+	precision: number;
+	recall: number;
+	f1: number;
+}> {
+	const data = await Deno.readTextFile("./data/filter/test1.jsonl");
+	const samples = data.split("\n")
+		.map((line) => {
+			try {
+				return JSON.parse(line);
+			} catch {
+				return null;
+			}
+		})
+		.filter(Boolean);
+
+	const allPredictions: number[] = [];
+	const allLabels: number[] = [];
+
+	const t = new Date().getTime();
+	for (const sample of samples) {
+		try {
+			const embeddings = await getONNXEmbeddings([
+				sample.title,
+				sample.description,
+				sample.tags.join(","),
+				sample.author_info,
+			], session);
+
+			const probabilities = await runClassification(embeddings);
+			allPredictions.push(probabilities.indexOf(Math.max(...probabilities)));
+			allLabels.push(sample.label);
+		} catch (error) {
+			console.error("Processing error:", error);
+		}
+	}
+	const elapsed = new Date().getTime() - t;
+
+	return calculateMetrics(allLabels, allPredictions, elapsed);
+}
+
+// 主函数
+async function main() {
+	await loadTokenizer();
+
+	// 评估原始模型
+	const originalMetrics = await evaluateModel(sessionEmbeddingOriginal);
+	console.log("Original Model Metrics:");
+	console.table(originalMetrics);
+
+	// 评估量化模型
+	const quantizedMetrics = await evaluateModel(sessionEmbeddingQuantized);
+	console.log("Quantized Model Metrics:");
+	console.table(quantizedMetrics);
+}
+
+await main();

packages/crawler/mq/exec/archiveSnapshots.ts

@@ -1,28 +1,11 @@
 import { Job } from "bullmq";
-import { getVideosWithoutActiveSnapshotScheduleByType, scheduleSnapshot } from "db/snapshotSchedule.ts";
+import { getAllVideosWithoutActiveSnapshotSchedule, scheduleSnapshot } from "db/snapshotSchedule.ts";
 import logger from "@core/log/logger.ts";
-import { lockManager } from "@core/mq/lockManager.ts";
+import { lockManager } from "mq/lockManager.ts";
 import { getLatestVideoSnapshot } from "db/snapshot.ts";
-import { MINUTE } from "@core/const/time.ts";
+import { HOUR, MINUTE } from "@core/const/time.ts";
 import { sql } from "@core/db/dbNew";
 
-function getNextSaturdayMidnightTimestamp(): number {
-	const now = new Date();
-	const currentDay = now.getDay();
-
-	let daysUntilNextSaturday = (6 - currentDay + 7) % 7;
-
-	if (daysUntilNextSaturday === 0) {
-		daysUntilNextSaturday = 7;
-	}
-
-	const nextSaturday = new Date(now);
-	nextSaturday.setDate(nextSaturday.getDate() + daysUntilNextSaturday);
-	nextSaturday.setHours(0, 0, 0, 0);
-
-	return nextSaturday.getTime();
-}
-
 export const archiveSnapshotsWorker = async (_job: Job) => {
 	try {
 		const startedAt = Date.now();
@@ -31,22 +14,21 @@ export const archiveSnapshotsWorker = async (_job: Job) => {
 			return;
 		}
 		await lockManager.acquireLock("dispatchArchiveSnapshots", 30 * 60);
-		const aids = await getVideosWithoutActiveSnapshotScheduleByType(sql, "archive");
+		const aids = await getAllVideosWithoutActiveSnapshotSchedule(sql);
 		for (const rawAid of aids) {
 			const aid = Number(rawAid);
 			const latestSnapshot = await getLatestVideoSnapshot(sql, aid);
 			const now = Date.now();
 			const lastSnapshotedAt = latestSnapshot?.time ?? now;
-			const nextSatMidnight = getNextSaturdayMidnightTimestamp();
+			const interval = 168;
-			const interval = nextSatMidnight - now;
 			logger.log(
 				`Scheduled archive snapshot for aid ${aid} in ${interval} hours.`,
 				"mq",
 				"fn:archiveSnapshotsWorker"
 			);
-			const targetTime = lastSnapshotedAt + interval;
+			const targetTime = lastSnapshotedAt + interval * HOUR;
 			await scheduleSnapshot(sql, aid, "archive", targetTime);
-			if (now - startedAt > 30 * MINUTE) {
+			if (now - startedAt > 250 * MINUTE) {
 				return;
 			}
 		}

packages/crawler/mq/exec/classifyVideo.ts

@@ -3,7 +3,7 @@ import { getUnlabelledVideos, getVideoInfoFromAllData, insertVideoLabel } from "
 import Akari from "ml/akari.ts";
 import { ClassifyVideoQueue } from "mq/index.ts";
 import logger from "@core/log/logger.ts";
-import { lockManager } from "@core/mq/lockManager.ts";
+import { lockManager } from "mq/lockManager.ts";
 import { aidExistsInSongs } from "db/songs.ts";
 import { insertIntoSongs } from "mq/task/collectSongs.ts";
 import { scheduleSnapshot } from "db/snapshotSchedule.ts";
@@ -34,7 +34,7 @@ export const classifyVideoWorker = async (job: Job) => {
 
 	await job.updateData({
 		...job.data,
-		label: label
+		label: label,
 	});
 
 	return 0;
@@ -46,19 +46,19 @@ export const classifyVideosWorker = async () => {
 		return;
 	}
 
-	await lockManager.acquireLock("classifyVideos", 5 * 60);
+	await lockManager.acquireLock("classifyVideos");
 
 	const videos = await getUnlabelledVideos(sql);
 	logger.log(`Found ${videos.length} unlabelled videos`);
 
-	const startTime = new Date().getTime();
+	let i = 0;
 	for (const aid of videos) {
-		const now = new Date().getTime();
+		if (i > 200) {
-		if (now - startTime > 4.2 * MINUTE) {
 			await lockManager.releaseLock("classifyVideos");
-			return 1;
+			return 10000 + i;
 		}
 		await ClassifyVideoQueue.add("classifyVideo", { aid: Number(aid) });
+		i++;
 	}
 	await lockManager.releaseLock("classifyVideos");
 	return 0;

packages/crawler/mq/exec/collectSongs.ts

@@ -1,7 +1,7 @@
 import { Job } from "bullmq";
 import { collectSongs } from "mq/task/collectSongs.ts";
 
-export const collectSongsWorker = async (_job: Job): Promise<void> => {
+export const collectSongsWorker = async (_job: Job): Promise<void> =>{
 	await collectSongs();
 	return;
-};
+}

packages/crawler/mq/exec/dispatchMilestoneSnapshots.ts

@@ -16,8 +16,8 @@ export const dispatchMilestoneSnapshotsWorker = async (_job: Job) => {
 			if (eta > 144) continue;
 			const now = Date.now();
 			const scheduledNextSnapshotDelay = eta * HOUR;
-			const maxInterval = 1.2 * HOUR;
+			const maxInterval = 1 * HOUR;
-			const minInterval = 2 * SECOND;
+			const minInterval = 1 * SECOND;
 			const delay = truncate(scheduledNextSnapshotDelay, minInterval, maxInterval);
 			const targetTime = now + delay;
 			await scheduleSnapshot(sql, aid, "milestone", targetTime);
@@ -25,5 +25,5 @@ export const dispatchMilestoneSnapshotsWorker = async (_job: Job) => {
 		}
 	} catch (e) {
 		logger.error(e as Error, "mq", "fn:dispatchMilestoneSnapshotsWorker");
-	}
+	};
-};
+}

packages/crawler/mq/exec/dispatchRegularSnapshots.ts

@@ -1,10 +1,10 @@
 import { Job } from "bullmq";
 import { getLatestVideoSnapshot } from "db/snapshot.ts";
 import { truncate } from "utils/truncate.ts";
-import { getVideosWithoutActiveSnapshotScheduleByType, scheduleSnapshot } from "db/snapshotSchedule.ts";
+import { getVideosWithoutActiveSnapshotSchedule, scheduleSnapshot } from "db/snapshotSchedule.ts";
 import logger from "@core/log/logger.ts";
 import { HOUR, MINUTE, WEEK } from "@core/const/time.ts";
-import { lockManager } from "@core/mq/lockManager.ts";
+import { lockManager } from "mq/lockManager.ts";
 import { getRegularSnapshotInterval } from "mq/task/regularSnapshotInterval.ts";
 import { sql } from "@core/db/dbNew.ts";
 
@@ -17,7 +17,7 @@ export const dispatchRegularSnapshotsWorker = async (_job: Job): Promise<void> =
 		}
 		await lockManager.acquireLock("dispatchRegularSnapshots", 30 * 60);
 
-		const aids = await getVideosWithoutActiveSnapshotScheduleByType(sql, "normal");
+		const aids = await getVideosWithoutActiveSnapshotSchedule(sql);
 		for (const rawAid of aids) {
 			const aid = Number(rawAid);
 			const latestSnapshot = await getLatestVideoSnapshot(sql, aid);

packages/crawler/mq/exec/executors.ts

@@ -7,4 +7,4 @@ export * from "./dispatchMilestoneSnapshots.ts";
 export * from "./dispatchRegularSnapshots.ts";
 export * from "./snapshotVideo.ts";
 export * from "./scheduleCleanup.ts";
-export * from "./snapshotTick.ts";
+export * from "./snapshotTick.ts";

packages/crawler/mq/exec/getLatestVideos.ts

@@ -2,6 +2,6 @@ import { sql } from "@core/db/dbNew";
 import { Job } from "bullmq";
 import { queueLatestVideos } from "mq/task/queueLatestVideo.ts";
 
-export const getLatestVideosWorker = async (_job: Job): Promise<void> => {
+export const getLatestVideosWorker = async (_job: Job): Promise<void> =>{
 	await queueLatestVideos(sql);
-};
+}

packages/crawler/mq/exec/getVideoInfo.ts

@@ -10,4 +10,4 @@ export const getVideoInfoWorker = async (job: Job): Promise<void> => {
 		return;
 	}
 	await insertVideoInfo(sql, aid);
-};
+}

packages/crawler/mq/exec/snapshotTick.ts

@@ -5,15 +5,15 @@ import {
 	getBulkSnapshotsInNextSecond,
 	getSnapshotsInNextSecond,
 	setSnapshotStatus,
-	videoHasProcessingSchedule
+	videoHasProcessingSchedule,
 } from "db/snapshotSchedule.ts";
 import logger from "@core/log/logger.ts";
 import { SnapshotQueue } from "mq/index.ts";
 import { sql } from "@core/db/dbNew";
 
 const priorityMap: { [key: string]: number } = {
-	milestone: 1,
+	"milestone": 1,
-	normal: 3
+	"normal": 3,
 };
 
 export const bulkSnapshotTickWorker = async (_job: Job) => {
@@ -35,16 +35,12 @@ export const bulkSnapshotTickWorker = async (_job: Job) => {
 					created_at: schedule.created_at,
 					started_at: schedule.started_at,
 					finished_at: schedule.finished_at,
-					status: schedule.status
+					status: schedule.status,
 				};
 			});
-			await SnapshotQueue.add(
+			await SnapshotQueue.add("bulkSnapshotVideo", {
-				"bulkSnapshotVideo",
+				schedules: schedulesData,
-				{
+			}, { priority: 3 });
-					schedules: schedulesData
-				},
-				{ priority: 3 }
-			);
 		}
 		return `OK`;
 	} catch (e) {
@@ -65,15 +61,11 @@ export const snapshotTickWorker = async (_job: Job) => {
 			}
 			const aid = Number(schedule.aid);
 			await setSnapshotStatus(sql, schedule.id, "processing");
-			await SnapshotQueue.add(
+			await SnapshotQueue.add("snapshotVideo", {
-				"snapshotVideo",
+				aid: Number(aid),
-				{
+				id: Number(schedule.id),
-					aid: Number(aid),
+				type: schedule.type ?? "normal",
-					id: Number(schedule.id),
+			}, { priority });
-					type: schedule.type ?? "normal"
-				},
-				{ priority }
-			);
 		}
 		return `OK`;
 	} catch (e) {
@@ -84,5 +76,5 @@ export const snapshotTickWorker = async (_job: Job) => {
 export const closetMilestone = (views: number) => {
 	if (views < 100000) return 100000;
 	if (views < 1000000) return 1000000;
-	return Math.ceil(views / 1000000) * 1000000;
+	return 10000000;
 };

packages/crawler/mq/exec/snapshotVideo.ts

@@ -1,19 +1,19 @@
 import { Job } from "bullmq";
-import { getLatestSnapshot, scheduleSnapshot, setSnapshotStatus, snapshotScheduleExists } from "db/snapshotSchedule.ts";
+import { scheduleSnapshot, setSnapshotStatus, snapshotScheduleExists } from "db/snapshotSchedule.ts";
 import logger from "@core/log/logger.ts";
 import { HOUR, MINUTE, SECOND } from "@core/const/time.ts";
+import { lockManager } from "mq/lockManager.ts";
 import { getBiliVideoStatus, setBiliVideoStatus } from "../../db/bilibili_metadata.ts";
 import { insertVideoSnapshot } from "mq/task/getVideoStats.ts";
 import { getSongsPublihsedAt } from "db/songs.ts";
 import { getAdjustedShortTermETA } from "mq/scheduling.ts";
 import { NetSchedulerError } from "@core/net/delegate.ts";
 import { sql } from "@core/db/dbNew.ts";
-import { closetMilestone } from "./snapshotTick.ts";
 
 const snapshotTypeToTaskMap: { [key: string]: string } = {
-	milestone: "snapshotMilestoneVideo",
+	"milestone": "snapshotMilestoneVideo",
-	normal: "snapshotVideo",
+	"normal": "snapshotVideo",
-	new: "snapshotMilestoneVideo"
+	"new": "snapshotMilestoneVideo",
 };
 
 export const snapshotVideoWorker = async (job: Job): Promise<void> => {
@@ -22,7 +22,6 @@ export const snapshotVideoWorker = async (job: Job): Promise<void> => {
 	const type = job.data.type;
 	const task = snapshotTypeToTaskMap[type] ?? "snapshotVideo";
 	const retryInterval = type === "milestone" ? 5 * SECOND : 2 * MINUTE;
-	const latestSnapshot = await getLatestSnapshot(sql, aid);
 	try {
 		const exists = await snapshotScheduleExists(sql, id);
 		if (!exists) {
@@ -33,7 +32,7 @@ export const snapshotVideoWorker = async (job: Job): Promise<void> => {
 			logger.warn(
 				`Video ${aid} has status ${status} in the database. Abort snapshoting.`,
 				"mq",
-				"fn:dispatchRegularSnapshotsWorker"
+				"fn:dispatchRegularSnapshotsWorker",
 			);
 			return;
 		}
@@ -45,7 +44,7 @@ export const snapshotVideoWorker = async (job: Job): Promise<void> => {
 			logger.warn(
 				`Bilibili return status ${status} when snapshoting for ${aid}.`,
 				"mq",
-				"fn:dispatchRegularSnapshotsWorker"
+				"fn:dispatchRegularSnapshotsWorker",
 			);
 			return;
 		}
@@ -53,7 +52,7 @@ export const snapshotVideoWorker = async (job: Job): Promise<void> => {
 		if (type === "new") {
 			const publihsedAt = await getSongsPublihsedAt(sql, aid);
 			const timeSincePublished = stat.time - publihsedAt!;
-			const viewsPerHour = (stat.views / timeSincePublished) * HOUR;
+			const viewsPerHour = stat.views / timeSincePublished * HOUR;
 			if (timeSincePublished > 48 * HOUR) {
 				return;
 			}
@@ -73,41 +72,46 @@ export const snapshotVideoWorker = async (job: Job): Promise<void> => {
 			await scheduleSnapshot(sql, aid, type, Date.now() + intervalMins * MINUTE, true);
 		}
 		if (type !== "milestone") return;
-		const alreadyAchievedMilestone = stat.views > closetMilestone(latestSnapshot.views);
-		if (alreadyAchievedMilestone) {
-			return;
-		}
 		const eta = await getAdjustedShortTermETA(sql, aid);
 		if (eta > 144) {
 			const etaHoursString = eta.toFixed(2) + " hrs";
 			logger.warn(
 				`ETA (${etaHoursString}) too long for milestone snapshot. aid: ${aid}.`,
 				"mq",
-				"fn:snapshotVideoWorker"
+				"fn:dispatchRegularSnapshotsWorker",
 			);
-			return;
 		}
 		const now = Date.now();
 		const targetTime = now + eta * HOUR;
 		await scheduleSnapshot(sql, aid, type, targetTime);
 		await setSnapshotStatus(sql, id, "completed");
 		return;
-	} catch (e) {
+	}
+	catch (e) {
 		if (e instanceof NetSchedulerError && e.code === "NO_PROXY_AVAILABLE") {
-			logger.warn(`No available proxy for aid ${job.data.aid}.`, "mq", "fn:snapshotVideoWorker");
+			logger.warn(
+				`No available proxy for aid ${job.data.aid}.`,
+				"mq",
+				"fn:takeSnapshotForVideoWorker",
+			);
 			await setSnapshotStatus(sql, id, "no_proxy");
-			await scheduleSnapshot(sql, aid, type, Date.now() + retryInterval, false, true);
+			await scheduleSnapshot(sql, aid, type, Date.now() + retryInterval);
 			return;
-		} else if (e instanceof NetSchedulerError && e.code === "ALICLOUD_PROXY_ERR") {
+		}
+		else if (e instanceof NetSchedulerError && e.code === "ALICLOUD_PROXY_ERR") {
 			logger.warn(
 				`Failed to proxy request for aid ${job.data.aid}: ${e.message}`,
 				"mq",
-				"fn:snapshotVideoWorker"
+				"fn:takeSnapshotForVideoWorker",
 			);
 			await setSnapshotStatus(sql, id, "failed");
 			await scheduleSnapshot(sql, aid, type, Date.now() + retryInterval);
 		}
-		logger.error(e as Error, "mq", "fn:snapshotVideoWorker");
+		logger.error(e as Error, "mq", "fn:takeSnapshotForVideoWorker");
 		await setSnapshotStatus(sql, id, "failed");
 	}
+	finally {
+		await lockManager.releaseLock("dispatchRegularSnapshots");
+	};
+	return;
 };

packages/crawler/mq/exec/takeBulkSnapshot.ts

@@ -3,7 +3,7 @@ import {
 	bulkScheduleSnapshot,
 	bulkSetSnapshotStatus,
 	scheduleSnapshot,
-	snapshotScheduleExists
+	snapshotScheduleExists,
 } from "db/snapshotSchedule.ts";
 import { bulkGetVideoStats } from "net/bulkGetVideoStats.ts";
 import logger from "@core/log/logger.ts";
@@ -46,16 +46,16 @@ export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
 			await sql`
                 INSERT INTO video_snapshot (aid, views, danmakus, replies, likes, coins, shares, favorites)
                 VALUES (
-					${aid},
+					${aid}, 
-					${views},
+					${views}, 
-					${danmakus},
+					${danmakus}, 
-					${replies},
+					${replies}, 
-					${likes},
+					${likes}, 
-					${coins},
+					${coins}, 
-					${shares},
+					${shares}, 
 					${favorites}
 				)
-			`;
+			`
 
 			logger.log(`Taken snapshot for video ${aid} in bulk.`, "net", "fn:takeBulkSnapshotForVideosWorker");
 		}
@@ -72,16 +72,13 @@ export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
 		return `DONE`;
 	} catch (e) {
 		if (e instanceof NetSchedulerError && e.code === "NO_PROXY_AVAILABLE") {
-			logger.warn(`No available proxy for bulk request now.`, "mq", "fn:takeBulkSnapshotForVideosWorker");
+			logger.warn(
-			await bulkSetSnapshotStatus(sql, ids, "no_proxy");
+				`No available proxy for bulk request now.`,
-			await bulkScheduleSnapshot(
+				"mq",
-				sql,
+				"fn:takeBulkSnapshotForVideosWorker",
-				aidsToFetch,
-				"normal",
-				Date.now() + 20 * MINUTE * Math.random(),
-				false,
-				true
 			);
+			await bulkSetSnapshotStatus(sql, ids, "no_proxy");
+			await bulkScheduleSnapshot(sql, aidsToFetch, "normal", Date.now() + 20 * MINUTE * Math.random());
 			return;
 		}
 		logger.error(e as Error, "mq", "fn:takeBulkSnapshotForVideosWorker");

packages/crawler/mq/index.ts

@@ -2,13 +2,13 @@ import { Queue, ConnectionOptions } from "bullmq";
 import { redis } from "@core/db/redis.ts";
 
 export const LatestVideosQueue = new Queue("latestVideos", {
-	connection: redis as ConnectionOptions
+    connection: redis as ConnectionOptions
 });
 
 export const ClassifyVideoQueue = new Queue("classifyVideo", {
-	connection: redis as ConnectionOptions
+    connection: redis as ConnectionOptions
 });
 
 export const SnapshotQueue = new Queue("snapshot", {
-	connection: redis as ConnectionOptions
+    connection: redis as ConnectionOptions
 });

packages/crawler/mq/init.ts

@@ -62,8 +62,8 @@ export async function initMQ() {
 	});
 
 	await SnapshotQueue.upsertJobScheduler("dispatchArchiveSnapshots", {
-		every: 2 * HOUR,
+		every: 6 * HOUR,
-		immediately: false
+		immediately: true
 	});
 
 	await SnapshotQueue.upsertJobScheduler("scheduleCleanup", {

packages/crawler/mq/lockManager.ts

@@ -1,5 +1,5 @@
 import { Redis } from "ioredis";
-import { redis } from "@core/db/redis.ts";
+import { redis } from "../../core/db/redis.ts";
 
 class LockManager {
 	private redis: Redis;

packages/crawler/mq/scheduling.ts

@@ -2,7 +2,7 @@ import { findClosestSnapshot, getLatestSnapshot, hasAtLeast2Snapshots } from "db
 import { truncate } from "utils/truncate.ts";
 import { closetMilestone } from "./exec/snapshotTick.ts";
 import { HOUR, MINUTE } from "@core/const/time.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 const log = (value: number, base: number = 10) => Math.log(value) / Math.log(base);
 
@@ -12,12 +12,13 @@ const getFactor = (x: number) => {
 	const c = 100;
 	const u = 0.601;
 	const g = 455;
-	if (x > g) {
+	if (x>g) {
-		return log(b / log(x + 1), a);
+		return log(b/log(x+1),a);
-	} else {
-		return log(b / log(x + c), a) + u;
 	}
-};
+	else {
+		return log(b/log(x+c),a)+u;
+	}
+}
 
 /*
  * Returns the minimum ETA in hours for the next snapshot
@@ -33,7 +34,7 @@ export const getAdjustedShortTermETA = async (sql: Psql, aid: number) => {
 	if (!snapshotsEnough) return 0;
 
 	const currentTimestamp = new Date().getTime();
-	const timeIntervals = [3 * MINUTE, 20 * MINUTE, HOUR, 3 * HOUR, 6 * HOUR, 72 * HOUR];
+	const timeIntervals = [3 * MINUTE, 20 * MINUTE, 1 * HOUR, 3 * HOUR, 6 * HOUR, 72 * HOUR];
 	const DELTA = 0.00001;
 	let minETAHours = Infinity;
 

packages/crawler/mq/task/collectSongs.ts

@@ -3,7 +3,7 @@ import { aidExistsInSongs, getNotCollectedSongs } from "db/songs.ts";
 import logger from "@core/log/logger.ts";
 import { scheduleSnapshot } from "db/snapshotSchedule.ts";
 import { MINUTE } from "@core/const/time.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 export async function collectSongs() {
 	const aids = await getNotCollectedSongs(sql);
@@ -25,5 +25,5 @@ export async function insertIntoSongs(sql: Psql, aid: number) {
 			(SELECT duration FROM bilibili_metadata WHERE aid = ${aid})
 		)
 		ON CONFLICT DO NOTHING
-	`;
+	`
 }

packages/crawler/mq/task/getVideoDetails.ts

@@ -4,7 +4,7 @@ import logger from "@core/log/logger.ts";
 import { ClassifyVideoQueue } from "mq/index.ts";
 import { userExistsInBiliUsers, videoExistsInAllData } from "../../db/bilibili_metadata.ts";
 import { HOUR, SECOND } from "@core/const/time.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 export async function insertVideoInfo(sql: Psql, aid: number) {
 	const videoExists = await videoExistsInAllData(sql, aid);
@@ -18,9 +18,9 @@ export async function insertVideoInfo(sql: Psql, aid: number) {
 	const bvid = data.View.bvid;
 	const desc = data.View.desc;
 	const uid = data.View.owner.mid;
-	const tags = data.Tags.filter((tag) => !["old_channel", "topic"].indexOf(tag.tag_type))
+	const tags = data.Tags
-		.map((tag) => tag.tag_name)
+		.filter((tag) => !["old_channel", "topic"].indexOf(tag.tag_type))
-		.join(",");
+		.map((tag) => tag.tag_name).join(",");
 	const title = data.View.title;
 	const published_at = formatTimestampToPsql(data.View.pubdate * SECOND + 8 * HOUR);
 	const duration = data.View.duration;
@@ -55,7 +55,7 @@ export async function insertVideoInfo(sql: Psql, aid: number) {
 			${stat.share}, 
 			${stat.favorite}
 		)
-	`;
+	`
 
 	logger.log(`Inserted video metadata for aid: ${aid}`, "mq");
 	await ClassifyVideoQueue.add("classifyVideo", { aid });

packages/crawler/mq/task/getVideoStats.ts

@@ -1,6 +1,6 @@
 import { getVideoInfo } from "@core/net/getVideoInfo.ts";
 import logger from "@core/log/logger.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 export interface SnapshotNumber {
 	time: number;
@@ -24,7 +24,11 @@ export interface SnapshotNumber {
  * - The native `fetch` function threw an error: with error code `FETCH_ERROR`
  * - The alicloud-fc threw an error: with error code `ALICLOUD_FC_ERROR`
  */
-export async function insertVideoSnapshot(sql: Psql, aid: number, task: string): Promise<number | SnapshotNumber> {
+export async function insertVideoSnapshot(
+	sql: Psql,
+	aid: number,
+	task: string,
+): Promise<number | SnapshotNumber> {
 	const data = await getVideoInfo(aid, task);
 	if (typeof data == "number") {
 		return data;
@@ -38,10 +42,10 @@ export async function insertVideoSnapshot(sql: Psql, aid: number, task: string):
 	const shares = data.stat.share;
 	const favorites = data.stat.favorite;
 
-	await sql`
+    await sql`
         INSERT INTO video_snapshot (aid, views, danmakus, replies, likes, coins, shares, favorites)
         VALUES (${aid}, ${views}, ${danmakus}, ${replies}, ${likes}, ${coins}, ${shares}, ${favorites})
-    `;
+    `
 
 	logger.log(`Taken snapshot for video ${aid}.`, "net", "fn:insertVideoSnapshot");
 
@@ -54,6 +58,6 @@ export async function insertVideoSnapshot(sql: Psql, aid: number, task: string):
 		coins,
 		shares,
 		favorites,
-		time
+		time,
 	};
 }

packages/crawler/mq/task/queueLatestVideo.ts

@@ -4,9 +4,11 @@ import { sleep } from "utils/sleep.ts";
 import { SECOND } from "@core/const/time.ts";
 import logger from "@core/log/logger.ts";
 import { LatestVideosQueue } from "mq/index.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
-export async function queueLatestVideos(sql: Psql): Promise<number | null> {
+export async function queueLatestVideos(
+	sql: Psql,
+): Promise<number | null> {
 	let page = 1;
 	let i = 0;
 	const videosFound = new Set();
@@ -24,18 +26,14 @@ export async function queueLatestVideos(sql: Psql): Promise<number | null> {
 			if (videoExists) {
 				continue;
 			}
-			await LatestVideosQueue.add(
+			await LatestVideosQueue.add("getVideoInfo", { aid }, {
-				"getVideoInfo",
+				delay,
-				{ aid },
+				attempts: 100,
-				{
+				backoff: {
-					delay,
+					type: "fixed",
-					attempts: 100,
+					delay: SECOND * 5,
-					backoff: {
+				},
-						type: "fixed",
+			});
-						delay: SECOND * 5
-					}
-				}
-			);
 			videosFound.add(aid);
 			allExists = false;
 			delay += Math.random() * SECOND * 1.5;
@@ -44,7 +42,7 @@ export async function queueLatestVideos(sql: Psql): Promise<number | null> {
 		logger.log(
 			`Page ${page} crawled, total: ${videosFound.size}/${i} videos added/observed.`,
 			"net",
-			"fn:queueLatestVideos()"
+			"fn:queueLatestVideos()",
 		);
 		if (allExists) {
 			return 0;

packages/crawler/mq/task/regularSnapshotInterval.ts

@@ -1,6 +1,6 @@
 import { findClosestSnapshot, findSnapshotBefore, getLatestSnapshot } from "db/snapshotSchedule.ts";
 import { HOUR } from "@core/const/time.ts";
-import type { Psql } from "@core/db/psql.d.ts";
+import type { Psql } from "global.d.ts";
 
 export const getRegularSnapshotInterval = async (sql: Psql, aid: number) => {
 	const now = Date.now();
@@ -14,7 +14,7 @@ export const getRegularSnapshotInterval = async (sql: Psql, aid: number) => {
 	if (hoursDiff < 8) return 24;
 	const viewsDiff = latestSnapshot.views - oldSnapshot.views;
 	if (viewsDiff === 0) return 72;
-	const speedPerDay = (viewsDiff / (hoursDiff + 0.001)) * 24;
+	const speedPerDay = viewsDiff / (hoursDiff + 0.001) * 24;
 	if (speedPerDay < 6) return 36;
 	if (speedPerDay < 120) return 24;
 	if (speedPerDay < 320) return 12;

packages/crawler/mq/task/removeAllTimeoutSchedules.ts

@@ -2,10 +2,14 @@ import { sql } from "@core/db/dbNew";
 import logger from "@core/log/logger.ts";
 
 export async function removeAllTimeoutSchedules() {
-	logger.log("Too many timeout schedules, directly removing these schedules...", "mq", "fn:scheduleCleanupWorker");
+	logger.log(
+		"Too many timeout schedules, directly removing these schedules...",
+		"mq",
+		"fn:scheduleCleanupWorker",
+	);
 	return await sql`
 		DELETE FROM snapshot_schedule
 		WHERE status IN ('pending', 'processing')
 		AND started_at < NOW() - INTERVAL '30 minutes'
 	`;
-}
+}

packages/crawler/package.json

@@ -1,6 +1,5 @@
 {
 	"name": "crawler",
-	"version": "1.3.0",
 	"scripts": {
 		"test": "bun --env-file=.env.test run vitest",
 		"worker:main": "bun run ./src/worker.ts",
@@ -8,8 +7,7 @@
 		"worker:filter": "bun run ./build/filterWorker.js",
 		"adder": "bun run ./src/jobAdder.ts",
 		"bullui": "bun run ./src/bullui.ts",
-		"all": "bun run concurrently --restart-tries -1 'bun run worker:main' 'bun run adder' 'bun run worker:filter'",
+		"all": "bun run concurrently --restart-tries -1 'bun run worker:main' 'bun run adder' 'bun run bullui' 'bun run worker:filter'"
-		"format": "prettier --write ."
 	},
 	"devDependencies": {
 		"concurrently": "^9.1.2"
@@ -21,7 +19,6 @@
 		"bullmq": "^5.52.1",
 		"express": "^5.1.0",
 		"ioredis": "^5.6.1",
-		"postgres": "^3.4.5",
 		"onnxruntime-node": "1.19.2"
 	}
 }