staff_data/packages/tus/src/server.ts

import http from "node:http";
import { EventEmitter } from "node:events";
import debug from "debug";
import { GetHandler } from "./handlers/GetHandler";
import { HeadHandler } from "./handlers/HeadHandler";
import { OptionsHandler } from "./handlers/OptionsHandler";
import { PatchHandler } from "./handlers/PatchHandler";
import { PostHandler } from "./handlers/PostHandler";
import { DeleteHandler } from "./handlers/DeleteHandler";
import { validateHeader } from "./validators/HeaderValidator";
import type stream from "node:stream";
import type { ServerOptions, RouteHandler, WithOptional } from "./types";
import { MemoryLocker } from "./lockers";
import {
	EVENTS,
	Upload,
	DataStore,
	REQUEST_METHODS,
	ERRORS,
	TUS_RESUMABLE,
	EXPOSED_HEADERS,
	CancellationContext,
} from "./utils";

/**
 * 处理器类型映射
 * 定义了TUS服务器支持的各种HTTP方法对应的处理器实例类型
 */
type Handlers = {
	GET: InstanceType<typeof GetHandler>; // GET请求处理器
	HEAD: InstanceType<typeof HeadHandler>; // HEAD请求处理器
	OPTIONS: InstanceType<typeof OptionsHandler>; // OPTIONS请求处理器
	PATCH: InstanceType<typeof PatchHandler>; // PATCH请求处理器
	POST: InstanceType<typeof PostHandler>; // POST请求处理器
	DELETE: InstanceType<typeof DeleteHandler>; // DELETE请求处理器
};

/**
 * TUS服务器事件接口定义
 * 描述了服务器在不同阶段触发的事件及其处理函数签名
 */
interface TusEvents {
	/**
	 * 文件创建后触发
	 * @param req HTTP请求对象
	 * @param res HTTP响应对象
	 * @param upload 上传对象实例
	 * @param url 生成的文件URL
	 */
	[EVENTS.POST_CREATE]: (
		req: http.IncomingMessage,
		res: http.ServerResponse,
		upload: Upload,
		url: string
	) => void;

	/**
	 * @deprecated 文件接收事件(已废弃)
	 * 建议使用 POST_RECEIVE_V2 替代
	 */
	[EVENTS.POST_RECEIVE]: (
		req: http.IncomingMessage,
		res: http.ServerResponse,
		upload: Upload
	) => void;

	/**
	 * 文件接收事件V2版本
	 * @param req HTTP请求对象
	 * @param upload 上传对象实例
	 */
	[EVENTS.POST_RECEIVE_V2]: (
		req: http.IncomingMessage,
		upload: Upload
	) => void;

	/**
	 * 文件上传完成事件
	 * @param req HTTP请求对象
	 * @param res HTTP响应对象
	 * @param upload 上传对象实例
	 */
	[EVENTS.POST_FINISH]: (
		req: http.IncomingMessage,
		res: http.ServerResponse,
		upload: Upload
	) => void;

	/**
	 * 文件终止上传事件
	 * @param req HTTP请求对象
	 * @param res HTTP响应对象
	 * @param id 文件唯一标识符
	 */
	[EVENTS.POST_TERMINATE]: (
		req: http.IncomingMessage,
		res: http.ServerResponse,
		id: string
	) => void;
}

/**
 * EventEmitter事件处理器类型别名
 */
type on = EventEmitter["on"];
type emit = EventEmitter["emit"];

/**
 * TUS服务器接口声明
 * 继承EventEmitter,支持事件监听和触发
 */
export declare interface Server {
	/**
	 * 为指定事件注册监听器
	 * @param event 事件名称，必须是TusEvents的键之一
	 * @param listener 事件触发时执行的回调函数
	 * @returns 返回Server实例以支持链式调用
	 */
	on<Event extends keyof TusEvents>(
		event: Event,
		listener: TusEvents[Event]
	): this;
	/**
	 * 为指定事件注册监听器（通用版本）
	 * @param eventName 事件名称
	 * @param listener 事件触发时执行的回调函数
	 * @returns 返回Server实例以支持链式调用
	 */
	on(eventName: Parameters<on>[0], listener: Parameters<on>[1]): this;
	/**
	 * 触发指定事件
	 * @param event 事件名称，必须是TusEvents的键之一
	 * @param listener 事件触发时执行的回调函数
	 * @returns 返回emit函数的返回值
	 */
	emit<Event extends keyof TusEvents>(
		event: Event,
		listener: TusEvents[Event]
	): ReturnType<emit>;
	/**
	 * 触发指定事件（通用版本）
	 * @param eventName 事件名称
	 * @param listener 事件触发时执行的回调函数
	 * @returns 返回emit函数的返回值
	 */
	emit(
		eventName: Parameters<emit>[0],
		listener: Parameters<emit>[1]
	): ReturnType<emit>;
}
/**
 * 调试日志工具实例
 */
const log = debug("tus-node-server");

// biome-ignore lint/suspicious/noUnsafeDeclarationMerging: it's fine
export class Server extends EventEmitter {
	datastore: DataStore;
	handlers: Handlers;
	options: ServerOptions;

	/**
	 * Server 构造函数
	 * @param options - 服务器配置选项，包含数据存储和可选配置
	 * @throws 如果未提供 options、path 或 datastore，将抛出错误
	 */
	constructor(
		options: WithOptional<ServerOptions, "locker"> & {
			datastore: DataStore;
		}
	) {
		super();

		if (!options) {
			throw new Error("'options' must be defined");
		}

		if (!options.path) {
			throw new Error("'path' is not defined; must have a path");
		}

		if (!options.datastore) {
			throw new Error(
				"'datastore' is not defined; must have a datastore"
			);
		}

		if (!options.locker) {
			options.locker = new MemoryLocker();
		}

		if (!options.lockDrainTimeout) {
			options.lockDrainTimeout = 3000;
		}

		if (!options.postReceiveInterval) {
			options.postReceiveInterval = 1000;
		}

		const { datastore, ...rest } = options;
		this.options = rest as ServerOptions;
		this.datastore = datastore;
		this.handlers = {
			// GET 请求处理器应在具体实现中编写
			GET: new GetHandler(this.datastore, this.options),
			// 这些方法按照 tus 协议处理
			HEAD: new HeadHandler(this.datastore, this.options),
			OPTIONS: new OptionsHandler(this.datastore, this.options),
			PATCH: new PatchHandler(this.datastore, this.options),
			POST: new PostHandler(this.datastore, this.options),
			DELETE: new DeleteHandler(this.datastore, this.options),
		};
		// 任何以方法为键分配给此对象的处理器将用于响应这些请求。
		// 当数据存储分配给服务器时，它们会被设置/重置。
		// 从服务器中移除任何事件监听器时，必须先从每个处理器中移除监听器。
		// 这必须在添加 'newListener' 监听器之前完成，以避免为所有请求处理器添加 'removeListener' 事件监听器。
		this.on("removeListener", (event: string, listener) => {
			this.datastore.removeListener(event, listener);
			for (const method of REQUEST_METHODS) {
				this.handlers[method].removeListener(event, listener);
			}
		});
		// 当事件监听器被添加到服务器时，确保它们从请求处理器冒泡到服务器级别。
		this.on("newListener", (event: string, listener) => {
			this.datastore.on(event, listener);
			for (const method of REQUEST_METHODS) {
				this.handlers[method].on(event, listener);
			}
		});
	}

	/**
	 * 注册 GET 请求处理器
	 * @param path - 请求路径
	 * @param handler - 请求处理器
	 */
	get(path: string, handler: RouteHandler) {
		this.handlers.GET.registerPath(path, handler);
	}

	/**
	 * 主服务器请求监听器，在每个 'request' 事件上调用
	 * @param req - HTTP 请求对象
	 * @param res - HTTP 响应对象
	 * @returns 返回 HTTP 响应对象或流
	 */
	async handle(
		req: http.IncomingMessage,
		res: http.ServerResponse
		// biome-ignore lint/suspicious/noConfusingVoidType: it's fine
	): Promise<http.ServerResponse | stream.Writable | void> {
		const context = this.createContext(req);
		log(`[TusServer] handle: ${req.method} ${req.url}`);
		// 允许覆盖 HTTP 方法。这样做的原因是某些库/环境不支持 PATCH 和 DELETE 请求，例如浏览器中的 Flash 和 Java 部分环境
		if (req.headers["x-http-method-override"]) {
			req.method = (
				req.headers["x-http-method-override"] as string
			).toUpperCase();
		}
		const onError = async (error: {
			status_code?: number;
			body?: string;
			message: string;
		}) => {
			let status_code =
				error.status_code || ERRORS.UNKNOWN_ERROR.status_code;
			let body =
				error.body ||
				`${ERRORS.UNKNOWN_ERROR.body}${error.message || ""}\n`;
			if (this.options.onResponseError) {
				const errorMapping = await this.options.onResponseError(
					req,
					res,
					error as Error
				);
				if (errorMapping) {
					status_code = errorMapping.status_code;
					body = errorMapping.body;
				}
			}
			return this.write(context, req, res, status_code, body);
		};
		if (req.method === "GET") {
			const handler = this.handlers.GET;
			return handler.send(req, res).catch(onError);
		}

		// Tus-Resumable 头部必须包含在每个请求和响应中，除了 OPTIONS 请求。其值必须是客户端或服务器使用的协议版本。
		res.setHeader("Tus-Resumable", TUS_RESUMABLE);
		if (
			req.method !== "OPTIONS" &&
			req.headers["tus-resumable"] === undefined
		) {
			return this.write(
				context,
				req,
				res,
				412,
				"Tus-Resumable Required\n"
			);
		}
		// 验证所有必需的头部以符合 tus 协议
		const invalid_headers = [];
		for (const header_name in req.headers) {
			if (req.method === "OPTIONS") {
				continue;
			}
			// 内容类型仅对 PATCH 请求进行检查。对于所有其他请求方法，它将被忽略并视为未设置内容类型，
			// 因为某些 HTTP 客户端可能会为此头部强制执行默认值。
			// 参见 https://github.com/tus/tus-node-server/pull/116
			if (
				header_name.toLowerCase() === "content-type" &&
				req.method !== "PATCH"
			) {
				continue;
			}
			if (
				!validateHeader(
					header_name,
					req.headers[header_name] as string | undefined
				)
			) {
				log(
					`Invalid ${header_name} header: ${req.headers[header_name]}`
				);
				invalid_headers.push(header_name);
			}
		}

		if (invalid_headers.length > 0) {
			return this.write(
				context,
				req,
				res,
				400,
				`Invalid ${invalid_headers.join(" ")}\n`
			);
		}
		// 启用 CORS
		res.setHeader("Access-Control-Allow-Origin", this.getCorsOrigin(req));
		res.setHeader("Access-Control-Expose-Headers", EXPOSED_HEADERS);
		if (this.options.allowedCredentials === true) {
			res.setHeader("Access-Control-Allow-Credentials", "true");
		}

		// 调用请求方法的处理器
		const handler = this.handlers[req.method as keyof Handlers];
		if (handler) {
			return handler.send(req, res, context).catch(onError);
		}

		return this.write(context, req, res, 404, "Not found\n");
	}

	/**
	 * 获取CORS（跨域资源共享）允许的源地址
	 *
	 * 该方法用于确定并返回允许的CORS源地址。首先检查请求头中的`origin`是否在允许的源列表中，
	 * 如果在则返回该`origin`；如果不在但允许的源列表不为空，则返回列表中的第一个源地址；
	 * 如果允许的源列表为空，则返回通配符`*`，表示允许所有源地址。
	 *
	 * @param req HTTP请求对象，包含请求头等信息
	 * @returns 返回允许的CORS源地址，可能是请求头中的`origin`、允许的源列表中的第一个源地址或通配符`*`
	 *
	 * 设计考量：
	 * - 该方法考虑了CORS策略的灵活性，允许通过配置动态指定允许的源地址。
	 * - 通过返回通配符`*`，简化了默认情况下的CORS配置，但需要注意这可能带来安全风险。
	 */
	private getCorsOrigin(req: http.IncomingMessage): string {
		const origin = req.headers.origin;
		// 检查请求头中的`origin`是否在允许的源列表中
		const isOriginAllowed =
			this.options.allowedOrigins?.some(
				(allowedOrigin) => allowedOrigin === origin
			) ?? true;
		// 如果`origin`存在且在允许的源列表中，则返回该`origin`
		if (origin && isOriginAllowed) {
			return origin;
		}
		// 如果允许的源列表不为空，则返回列表中的第一个源地址
		if (
			this.options.allowedOrigins &&
			this.options.allowedOrigins.length > 0
		) {
			return this.options.allowedOrigins[0];
		}

		// 如果允许的源列表为空，则返回通配符`*`，表示允许所有源地址
		return "*";
	}

	/**
	 * 写入响应
	 * @param context - 取消上下文
	 * @param req - HTTP 请求对象
	 * @param res - HTTP 响应对象
	 * @param status - HTTP 状态码
	 * @param body - 响应体
	 * @param headers - 响应头部
	 * @returns 返回 HTTP 响应对象
	 */
	write(
		context: CancellationContext,
		req: http.IncomingMessage,
		res: http.ServerResponse,
		status: number,
		body = "",
		headers = {}
	) {
		const isAborted = context.signal.aborted;

		if (status !== 204) {
			// @ts-expect-error not explicitly typed but possible
			headers["Content-Length"] = Buffer.byteLength(body, "utf8");
		}

		if (isAborted) {
			// 此条件处理请求被标记为中止的情况。
			// 在这种情况下，服务器通知客户端连接将被关闭。
			// 这是通过在响应中设置 'Connection' 头部为 'close' 来传达的。
			// 这一步对于防止服务器继续处理不再需要的请求至关重要，从而节省资源。

			// @ts-expect-error not explicitly typed but possible
			headers.Connection = "close";

			// 为响应 ('res') 添加 'finish' 事件的事件监听器。
			// 'finish' 事件在响应已发送给客户端时触发。
			// 一旦响应完成，请求 ('req') 对象将被销毁。
			// 销毁请求对象是释放与此请求相关的任何资源的关键步骤，因为它已经被中止。
			res.on("finish", () => {
				req.destroy();
			});
		}

		res.writeHead(status, headers);
		res.write(body);
		return res.end();
	}

	/**
	 * 启动服务器监听
	 * @param args - 监听参数
	 * @returns 返回 HTTP 服务器实例
	 */
	// biome-ignore lint/suspicious/noExplicitAny: todo
	listen(...args: any[]): http.Server {
		return http.createServer(this.handle.bind(this)).listen(...args);
	}

	/**
	 * 清理过期的上传
	 * @returns 返回删除的过期上传数量
	 * @throws 如果数据存储不支持过期扩展，将抛出错误
	 */
	cleanUpExpiredUploads(): Promise<number> {
		if (!this.datastore.hasExtension("expiration")) {
			throw ERRORS.UNSUPPORTED_EXPIRATION_EXTENSION;
		}

		return this.datastore.deleteExpired();
	}

	/**
	 * 创建取消上下文
	 * @param req - HTTP 请求对象
	 * @returns 返回取消上下文
	 */
	protected createContext(req: http.IncomingMessage) {
		// 初始化两个 AbortController：
		// 1. `requestAbortController` 用于即时请求终止，特别适用于在发生错误时停止客户端上传。
		// 2. `abortWithDelayController` 用于在终止前引入延迟，允许服务器有时间完成正在进行的操作。
		// 这在未来的请求可能需要获取当前请求持有的锁时特别有用。
		const requestAbortController = new AbortController();
		const abortWithDelayController = new AbortController();

		// 当 `abortWithDelayController` 被触发时调用此函数，以在指定延迟后中止请求。
		const onDelayedAbort = (err: unknown) => {
			abortWithDelayController.signal.removeEventListener(
				"abort",
				onDelayedAbort
			);
			setTimeout(() => {
				requestAbortController.abort(err);
			}, this.options.lockDrainTimeout);
		};
		abortWithDelayController.signal.addEventListener(
			"abort",
			onDelayedAbort
		);

		// 当请求关闭时，移除监听器以避免内存泄漏。
		req.on("close", () => {
			abortWithDelayController.signal.removeEventListener(
				"abort",
				onDelayedAbort
			);
		});

		// 返回一个对象，包含信号和两个中止请求的方法。
		// `signal` 用于监听请求中止事件。
		// `abort` 方法用于立即中止请求。
		// `cancel` 方法用于启动延迟中止序列。
		return {
			signal: requestAbortController.signal,
			abort: () => {
				// 立即中止请求
				if (!requestAbortController.signal.aborted) {
					requestAbortController.abort(ERRORS.ABORTED);
				}
			},
			cancel: () => {
				// 启动延迟中止序列，除非它已经在进行中。
				if (!abortWithDelayController.signal.aborted) {
					abortWithDelayController.abort(ERRORS.ABORTED);
				}
			},
		};
	}
}