固定 Java 21 release 编译并更新项目说明

安全修复：
- netty-all 4.1.101.Final -> 4.1.138.Final：4.1.101 存在多个已知 CVE
  （netty-codec-http 修复线 4.1.108/4.1.125/4.1.129/4.1.132-4.1.137，
  netty-common 修复线 4.1.115/4.1.118）。该组件直接面向主站与公网。
- commons-compress 1.25.0 -> 1.28.0（CVE-2024-26308/25710 修复于 1.26.0）。
- logback-classic 1.4.14 -> 1.5.38，slf4j 2.0.9 -> 2.0.19。
- jackson-databind 2.15.2 -> 2.22.2：留在 Jackson 2.x 线而非迁 3，因为本组件
  负责解析并落盘快照数据，且与主站（已升 Jackson 3）的跨版本线上格式已实测兼容。
- hutool 5.8.26 -> 5.8.47；lombok 1.18.40 -> 1.18.48；junit 5.10.2 -> 5.14.4。
- native-maven-plugin 0.10.3 -> 1.1.8（配合 GraalVM 25）。

移除 org.apache.httpcomponents:httpclient 4.5.14：全仓库无任何引用
（HTTP 调用走 hutool 的 HttpRequest），且该线自 2022-11 起已 EOL。

验证：mvn test 5 项全过；隔离冒烟加载线上真实快照目录（53 个文件）——
Jackson 2.22.2 正确解析由 2.15.2 写出的 manifest，/health/subscription 报
state=ready、12 账号/12 绑定，/sub/v2 与 /sub/cat 分别 43024/125045 字节
（与升级前完全一致）；与 Boot 4 主站（Jackson 3）端到端联调，
订阅快照 result=0（APPLY_SUCCESS）。

.gitignore

@@ -36,3 +36,7 @@ build/
 
 ### Mac OS ###
 .DS_Store
+/.idea/encodings.xml
+
+# 本地运行与测试日志
+run.out

.idea/encodings.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Encoding">
+    <file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
+  </component>
+</project>

.idea/misc.xml

@@ -8,7 +8,11 @@
       </list>
     </option>
   </component>
-  <component name="ProjectRootManager" version="2" languageLevel="JDK_17" default="true" project-jdk-name="17" project-jdk-type="JavaSDK">
+  <component name="PWA">
+    <option name="enabled" value="true" />
+    <option name="wasEnabledAtLeastOnce" value="true" />
+  </component>
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_21" default="true" project-jdk-name="21" project-jdk-type="JavaSDK">
     <output url="file://$PROJECT_DIR$/out" />
   </component>
 </project>

llm_readme.txt

@@ -0,0 +1,165 @@
+# storageNode 项目结构
+
+## 概述
+分布式存储/下载节点服务端，用于画廊网站。职责包括：从远程源下载图片集、压缩为 ZIP 归档、通过 HTTP 提供下载（支持断点续传）、管理代理订阅配置（V2Ray/Clash），并通过 Netty 自定义 TCP 协议与中心服务器通信。
+
+- **Group ID:** `org.lion`
+- **Version:** `1.0`
+- **Java 版本:** 21 字节码（生产使用 GraalVM JDK 25）
+- **构建工具:** Maven（单模块）
+- **编译目标:** 当前以 JVM/JAR + `lib/` 运行；保留 GraalVM 原生配置但尚未完成 JDK 25 原生验证
+
+---
+
+## 目录结构
+
+```
+storageNode/
+├── pom.xml                                              # Maven 构建配置
+└── src/
+    ├── main/
+    │   ├── java/
+    │   │   └── lion/
+    │   │       ├── Main.java                            # 程序入口
+    │   │       ├── storageNode.java                     # Netty TCP 核心节点
+    │   │       ├── CustomUtil.java                      # 工具方法（HTTP 通知、端口查找、ObjectMapper）
+    │   │       ├── MultiThreadedHTTPServer.java          # HTTP 文件下载服务（端口 8888）
+    │   │       ├── Config/
+    │   │       │   └── Config.java                      # 加载 config.properties
+    │   │       ├── Domain/
+    │   │       │   └── GalleryTask.java                  # 下载任务领域模型
+    │   │       ├── ErrorCode/
+    │   │       │   └── ErrorCode.java                   # 错误码常量
+    │   │       ├── Externel/
+    │   │       │   └── BackupSubServer.java             # 代理订阅文件服务（端口 8889）
+    │   │       ├── Message/
+    │   │       │   ├── AbstractMessage.java             # 消息基类 + 消息类型常量
+    │   │       │   ├── MessageCodec.java                # Netty 编解码器（ByteBuf ↔ AbstractMessage，JSON 格式）
+    │   │       │   └── Main/
+    │   │       │       ├── AvailableCheckMessage.java   # 可用性检查（type=8）
+    │   │       │       ├── DeleteGalleryMessage.java    # 删除画廊请求（type=3）
+    │   │       │       ├── DownloadPostMessage.java     # 提交下载任务（type=1）
+    │   │       │       ├── DownloadStatusMessage.java   # 上报任务状态（type=2）
+    │   │       │       ├── IdentityMessage.java         # 身份认证握手（type=6）
+    │   │       │       ├── MaintainMessage.java         # 心跳维持（type=7）
+    │   │       │       └── ResponseMessage.java         # 通用响应（type=0）
+    │   │       └── Service/
+    │   │           ├── DeleteService.java               # 删除画廊目录
+    │   │           └── DownloadCheckService.java        # 下载监控与压缩服务
+    │   └── resources/
+    │       ├── config.properties                        # DouNai 订阅地址配置
+    │       ├── simplelogger.properties                  # SLF4J 日志配置（输出到 run.out）
+    │       └── reflect-config.json                      # GraalVM 反射配置（Jackson 序列化）
+    └── test/
+        └── java/                                        # 订阅快照与下载/压缩恢复测试
+```
+
+---
+
+## 关键依赖
+
+| 依赖 | 版本 | 用途 |
+|---|---|---|
+| `io.netty:netty-all` | 4.1.138.Final | TCP 服务端/客户端 |
+| `com.fasterxml.jackson.core:jackson-databind` | 2.22.2 | JSON 序列化 |
+| `org.projectlombok:lombok` | 1.18.48 | 简化样板代码（`@Data`, `@Slf4j`） |
+| `ch.qos.logback:logback-classic` | 1.5.38 | 日志实现 |
+| `cn.hutool:hutool-all` | 5.8.47 | 文件操作、ZIP 压缩、HTTP 请求 |
+| `org.apache.commons:commons-compress` | 1.28.0 | 压缩归档 |
+| `org.graalvm.buildtools:native-maven-plugin` | 1.1.8 | 原生镜像配置（待验证） |
+
+---
+
+## 启动流程
+
+1. `Main.main()` → 调用 `boot()`（遗留的 Netty Bootstrap），然后执行 `Config.loadConfig()`
+2. 在新线程中启动 `BackupSubServer`（端口 8889）— 提供代理订阅文件下载
+3. 在新线程中启动 `MultiThreadedHTTPServer`（端口 8888）— 提供画廊 ZIP 文件下载
+4. 主线程创建 `storageNode()` 实例（阻塞构造函数，永不返回）：
+   - 从端口 26321 开始查找空闲端口，绑定 Netty TCP 服务端
+   - 作为 TCP 客户端连接 `lionwebsite.xyz:26322~26342`，唤醒中心服务器
+   - 启动 `DownloadCheckService` 和 5 秒定时任务 `mainThread()`
+
+---
+
+## 通信协议（Netty TCP）
+
+自定义协议格式：`[1字节类型] + [4字节长度] + [JSON 负载]`
+
+| 类型字节 | 消息类 | 方向 |
+|---|---|---|
+| 0 | `ResponseMessage` | 响应 |
+| 1 | `DownloadPostMessage` | 服务端 → 节点 |
+| 2 | `DownloadStatusMessage` | 节点 → 服务端 |
+| 3 | `DeleteGalleryMessage` | 服务端 → 节点 |
+| 6 | `IdentityMessage` | 握手 |
+| 7 | `MaintainMessage` | 心跳 |
+| 8 | `AvailableCheckMessage` | 可用性检查 |
+| 9 | `SubscriptionSnapshotMessage` | 主站 → 节点，完整订阅备机快照 |
+
+---
+
+## HTTP 服务
+
+### MultiThreadedHTTPServer（端口 8888）
+- 提供压缩后的画廊 ZIP 文件下载
+- 仅接受来自 `lionwebsite.xyz` IP 的连接
+- 支持 HTTP 206 Partial Content（断点续传）
+- 参数：`AuthCode`（管理员访问）、`gid`（画廊 ID）
+
+### BackupSubServer（端口 8889）
+- 提供 V2Ray 和 Clash 代理订阅文件
+- 不直接访问上游；主站完成下载和倍率过滤后，通过 Netty 类型 9 推送完整快照
+- 按公开 Key 的 SHA-256 查找用户绑定的子账号，未知 Key 返回 404
+- 快照经 SHA-256 和 HMAC 校验后原子落盘，主站离线时继续分发最后成功版本
+- 文件存储于 `sub/snapshots/{revision}/accounts/{accountId}/`
+- 没有有效快照或快照超过最大有效期时返回 503，不回退旧共享订阅
+- `GET /health/subscription` 提供不含 Key 和订阅正文的快照状态
+
+---
+
+## 核心服务
+
+### DownloadCheckService
+- 扫描下载目录（`/root/gallery/hentai/download/`）监控进度
+- 通过检测 `galleryinfo.txt` 文件判断下载完成
+- 将完成的下载任务移入压缩队列
+- 后台线程每 5 秒执行 ZIP 压缩，完成后删除源目录
+
+### DeleteService
+- 按名称删除画廊目录
+- 失败时返回 `ErrorCode.IO_ERROR` 或 `ErrorCode.FILE_NOT_FOUND`
+
+---
+
+## 配置说明
+
+### config.properties（从 `/root/gallery/storageNode/config.properties` 加载）
+```properties
+SubscriptionSyncEnabled=false
+SubscriptionSyncSecret=
+SubscriptionDataDir=/root/gallery/storageNode/sub
+SubscriptionMaxStaleSeconds=604800
+SubscriptionMaxPayloadBytes=52428800
+SubscriptionHttpPort=8889
+SubscriptionHttpWorkers=4
+SubscriptionSocketTimeoutMs=10000
+```
+
+生产同步密钥优先通过 `SUBSCRIPTION_SYNC_SECRET` 环境变量提供，不得提交到仓库或写入日志。
+
+### simplelogger.properties
+- 日志级别：`info`
+- 时间戳格式：`yyyy-MM-dd HH:mm:ss`
+- 输出文件：`run.out`
+
+---
+
+## 架构说明
+
+- 单模块 Maven 项目，现有 5 个测试用例覆盖订阅快照和压缩失败恢复
+- 硬编码文件系统路径（`/root/gallery/...`）→ 仅限 Linux 部署
+- 外部连接：`lionwebsite.xyz`、`personal.lionwebsite.xyz`、`aaaa.gay`
+- GraalVM 原生镜像编译，包含 Jackson 反射配置
+- 大量使用 Lombok（`@Data`、`@Slf4j`）
+- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作

pom.xml

@@ -9,8 +9,7 @@
     <version>1.0</version>
 
     <properties>
-        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.release>21</maven.compiler.release>
-        <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
@@ -18,43 +17,52 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>
-            <version>4.1.86.Final</version>
+            <version>4.1.138.Final</version>
         </dependency>
 
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.15.2</version>
+            <version>2.22.2</version>
         </dependency>
 
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
-            <version>1.18.28</version>
+            <version>1.18.48</version>
         </dependency>
 
         <dependency>
-            <groupId>org.im4java</groupId>
+            <groupId>org.slf4j</groupId>
-            <artifactId>im4java</artifactId>
+            <artifactId>slf4j-api</artifactId>
-            <version>1.4.0</version>
+            <version>2.0.19</version>
         </dependency>
 
+
         <dependency>
-            <groupId>log4j</groupId>
+            <groupId>ch.qos.logback</groupId>
-            <artifactId>log4j</artifactId>
+            <artifactId>logback-classic</artifactId>
-            <version>1.2.17</version>
+            <version>1.5.38</version>
         </dependency>
 
+
         <dependency>
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
-            <version>5.8.18</version>
+            <version>5.8.47</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
-            <version>1.21</version>
+            <version>1.28.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.14.4</version>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
@@ -62,25 +70,64 @@
         <plugins>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
+                <artifactId>maven-surefire-plugin</artifactId>
-                <version>3.4.2</version>
+                <version>3.2.5</version>
-                <executions>
+            </plugin>
-                    <execution>
+            <plugin>
-                        <phase>package</phase>
+                <groupId>org.apache.maven.plugins</groupId>
-                        <goals>
+                <artifactId>maven-compiler-plugin</artifactId>
-                            <goal>single</goal>
+                <version>3.13.0</version>
-                        </goals>
+                <configuration>
-                    </execution>
+                    <annotationProcessorPaths>
-                </executions>
+                        <path>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok</artifactId>
+                            <version>1.18.48</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+<!--            <plugin>-->
+<!--                <groupId>org.apache.maven.plugins</groupId>-->
+<!--                <artifactId>maven-assembly-plugin</artifactId>-->
+<!--                <version>3.4.2</version>-->
+<!--                <executions>-->
+<!--                    <execution>-->
+<!--                        <phase>package</phase>-->
+<!--                        <goals>-->
+<!--                            <goal>single</goal>-->
+<!--                        </goals>-->
+<!--                    </execution>-->
+<!--                </executions>-->
+<!--                <configuration>-->
+<!--                    <descriptorRefs>-->
+<!--                        <descriptorRef>jar-with-dependencies</descriptorRef>-->
+<!--                    </descriptorRefs>-->
+<!--                    <archive>-->
+<!--                        <manifest>-->
+<!--                            <mainClass>lion.Main</mainClass>-->
+<!--                        </manifest>-->
+<!--                    </archive>-->
+<!--                </configuration>-->
+<!--            </plugin>-->
+            <plugin>
+                <groupId>org.graalvm.buildtools</groupId>
+                <artifactId>native-maven-plugin</artifactId>
+                <version>1.1.8</version>
                 <configuration>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                    <archive>
-                        <manifest>
                     <mainClass>lion.Main</mainClass>
-                        </manifest>
+                    <imageName>storageNode</imageName>
-                    </archive>
+                    <buildArgs>
+                        <arg>-H:+ReportExceptionStackTraces</arg>
+                        <arg>--enable-url-protocols=https</arg>
+                        <arg>-H:IncludeResources="logback.xml"</arg>
+                        <arg>--initialize-at-build-time=ch.qos.logback.classic,ch.qos.logback.core,ch.qos.logback.classic.pattern,ch.qos.logback.core.pattern</arg>
+                        <arg>-H:ReflectionConfigurationFiles=src/main/resources/reflect-config.json</arg>
+                    </buildArgs>
+                    <metadataRepository>
+                        <enabled>true</enabled>
+                    </metadataRepository>
+
                 </configuration>
             </plugin>
         </plugins>

src/main/java/lion/Config/Config.java

@@ -0,0 +1,45 @@
+package lion.Config;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Config {
+    public static boolean subscriptionSyncEnabled;
+    public static String subscriptionSyncSecret;
+    public static String subscriptionDataDir;
+    public static long subscriptionMaxStaleSeconds;
+    public static int subscriptionMaxPayloadBytes;
+    public static int subscriptionHttpPort;
+    public static int subscriptionHttpWorkers;
+    public static int subscriptionSocketTimeoutMs;
+
+    public static void loadConfig(){
+        Properties prop = new Properties();
+
+        try (InputStream input = new FileInputStream("/root/gallery/storageNode/config.properties")) {
+            prop.load(input);
+            subscriptionSyncEnabled = Boolean.parseBoolean(value(prop, "SubscriptionSyncEnabled", "false"));
+            subscriptionSyncSecret = System.getenv().getOrDefault("SUBSCRIPTION_SYNC_SECRET",
+                    value(prop, "SubscriptionSyncSecret", ""));
+            subscriptionDataDir = value(prop, "SubscriptionDataDir", "/root/gallery/storageNode/sub");
+            subscriptionMaxStaleSeconds = Long.parseLong(value(prop, "SubscriptionMaxStaleSeconds", "604800"));
+            subscriptionMaxPayloadBytes = Integer.parseInt(value(prop, "SubscriptionMaxPayloadBytes", "52428800"));
+            subscriptionHttpPort = Integer.parseInt(value(prop, "SubscriptionHttpPort", "8889"));
+            subscriptionHttpWorkers = Integer.parseInt(value(prop, "SubscriptionHttpWorkers", "4"));
+            subscriptionSocketTimeoutMs = Integer.parseInt(value(prop, "SubscriptionSocketTimeoutMs", "10000"));
+            if (subscriptionSyncEnabled && subscriptionSyncSecret.isBlank())
+                throw new IllegalStateException("启用订阅同步时必须配置 SUBSCRIPTION_SYNC_SECRET");
+        } catch (IOException ex) {
+            log.error("加载配置失败:{}", ex.getMessage());
+        }
+    }
+
+    private static String value(Properties prop, String key, String fallback) {
+        return prop.getProperty(key, fallback).trim();
+    }
+}

src/main/java/lion/CustomUtil.java

@@ -0,0 +1,94 @@
+package lion;
+
+import cn.hutool.http.HttpRequest;
+import cn.hutool.http.HttpResponse;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.*;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+
+@Slf4j
+public class CustomUtil {
+
+    public static ObjectMapper objectMapper = new ObjectMapper();
+
+    public static void notifyMe(String message) {
+        String url = "https://personal.lionwebsite.xyz/message2me?AuthCode=alone&message=" + message;
+        HttpRequest request = HttpRequest.post(url);
+        request.header("User-Agent", "Mozilla/5.0");
+        try(HttpResponse response = request.execute()) {
+            if(response.getStatus() != 200) {
+                System.out.println("通知失败, status code:" + response.getStatus() + ", message:" + message);
+            }
+        }
+    }
+
+    public static int _findIdlePort(int port) {
+        for(int i=port; i<65535; i++){
+            try(ServerSocket ignored = new ServerSocket(i)){
+                return i;
+            }catch (IOException ignored) {
+            }
+        }
+        return -1;
+    }
+
+    public static String getRequestHeader(BufferedReader requestReader) throws IOException {
+        String line;
+        while ((line = requestReader.readLine()) != null) {
+            if (line.trim().isEmpty()) {
+                break;
+            }
+            if (line.startsWith("Range:")) {
+                return line.substring("Range".length() + 1).trim();
+            }
+        }
+        return null;
+    }
+
+    public static void sendErrorResponse(Socket clientSocket, String statusCode) throws IOException {
+        OutputStream responseStream = clientSocket.getOutputStream();
+        PrintWriter responseWriter = new PrintWriter(responseStream, true);
+        responseWriter.println("HTTP/1.1 " + statusCode);
+        responseWriter.println("Content-Type: text/html");
+        responseWriter.println();
+        responseWriter.println("<h1>" + statusCode + "</h1>");
+        responseStream.close();
+    }
+
+    public static void sendFileRange(Socket clientSocket, File file, long startByte, long endByte) throws IOException {
+        sendFileRange(clientSocket, file, startByte, endByte, null);
+    }
+
+    public static void sendFileRange(Socket clientSocket, File file, long startByte, long endByte, String contentDispositionFileName) throws IOException {
+        long fileLength = file.length();
+        OutputStream responseStream = clientSocket.getOutputStream();
+        PrintWriter responseWriter = new PrintWriter(responseStream, true);
+        responseWriter.println("HTTP/1.1 206 Partial Content");
+        responseWriter.println("Content-Type: application/octet-stream");
+        responseWriter.println("Accept-Ranges: bytes");
+        responseWriter.println("Content-Length: " + (endByte - startByte + 1));
+        responseWriter.println("Content-Range: bytes " + startByte + "-" + endByte + "/" + fileLength);
+        if (contentDispositionFileName != null) {
+            responseWriter.println("Content-Disposition: attachment; filename=\"" + contentDispositionFileName + "\"");
+        }
+        responseWriter.println();
+
+        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
+            randomAccessFile.seek(startByte);
+            byte[] buffer = new byte[1024];
+            int bytesRead;
+            long bytesRemaining = endByte - startByte + 1;
+            while (bytesRemaining > 0 && (bytesRead = randomAccessFile.read(buffer, 0, (int) Math.min(buffer.length, bytesRemaining))) != -1) {
+                responseStream.write(buffer, 0, bytesRead);
+                bytesRemaining -= bytesRead;
+            }
+        } catch (SocketException ignore) {
+        } finally {
+            responseStream.close();
+        }
+    }
+}

src/main/java/lion/Domain/GalleryTask.java

@@ -6,37 +6,22 @@ import lombok.Data;
 
 @Data
 public class GalleryTask {
-
+    public static final byte DOWNLOADING = 1;
-    public static byte DOWNLOADING = 1;
+    public static final byte DOWNLOAD_COMPLETE = 2;
-
+    public static final byte COMPRESSING = 3;
-    public static byte DOWNLOAD_COMPLETE = 2;
+    public static final byte COMPRESS_COMPLETE = 4;
-
-    public static byte DOWNLOAD_QUEUED = 3;
-
-    public static byte COMPRESS_COMPLETE = 4;
-
-
-    public static byte DOWNLOAD_ALL = 3;
-    public static byte DOWNLOAD_PREVIEW = 2;
-    public static byte DOWNLOAD_SOURCE = 1;
-
-
 
     @JsonInclude(JsonInclude.Include.NON_NULL)
-    private String name;
+    private volatile String name;
 
     private int gid;
 
-    private int pages;
+    private volatile byte status;
 
-    private byte status;
+    private volatile int proceeding;
-
-    private int proceeding;
-
-    private byte type;
 
     @JsonIgnore
-    private String path;
+    private volatile String path;
 
     @JsonIgnore
     public boolean is_download_complete(){
@@ -47,4 +32,9 @@ public class GalleryTask {
     public boolean is_compress_complete(){
         return status == COMPRESS_COMPLETE;
     }
+
+    @JsonIgnore
+    public boolean is_compressing(){
+        return status == COMPRESSING;
+    }
 }

src/main/java/lion/ErrorCode/ErrorCode.java

@@ -3,7 +3,4 @@ package lion.ErrorCode;
 public class ErrorCode {
     public static final byte IO_ERROR = 1;
     public static final byte FILE_NOT_FOUND = 2;
-
-    public static final byte COMPRESS_ERROR = 3;
-
 }

src/main/java/lion/Externel/BackupSubServer.java

@@ -0,0 +1,159 @@
+package lion.Externel;
+
+import lion.Config.Config;
+import lion.CustomUtil;
+import lion.Service.SubscriptionSnapshotStore;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.*;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.*;
+
+/** HTTP distributor for the last-known-good subscription snapshot. */
+@Slf4j
+public final class BackupSubServer implements Runnable {
+    private final SubscriptionSnapshotStore snapshotStore;
+    private final int port;
+    private final ExecutorService workers;
+
+    public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) {
+        this.snapshotStore = Objects.requireNonNull(snapshotStore);
+        this.port = port;
+        int workersCount = Math.max(1, workerCount);
+        this.workers = new ThreadPoolExecutor(workersCount, workersCount, 0, TimeUnit.MILLISECONDS,
+                new ArrayBlockingQueue<>(workersCount * 32), new ThreadPoolExecutor.AbortPolicy());
+    }
+
+    @Override
+    public void run() {
+        try (ServerSocket serverSocket = new ServerSocket(port)) {
+            log.info("备机订阅服务监听端口 {}", port);
+            while (!Thread.currentThread().isInterrupted()) {
+                Socket socket = serverSocket.accept();
+                try {
+                    workers.execute(() -> handle(socket));
+                } catch (RejectedExecutionException e) {
+                    try (socket) {
+                        send(socket, 503, "Service Unavailable", "text/plain", new byte[0], false);
+                    } catch (IOException ignored) { }
+                }
+            }
+        } catch (IOException e) {
+            log.error("备机订阅服务停止: {}", e.getMessage());
+        } finally {
+            workers.shutdownNow();
+        }
+    }
+
+    private void handle(Socket socket) {
+        try (socket) {
+            socket.setSoTimeout(Config.subscriptionSocketTimeoutMs);
+            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII));
+            String requestLine = reader.readLine();
+            if (requestLine == null || requestLine.length() > 2048) {
+                send(socket, 400, "Bad Request", "text/plain", new byte[0], false);
+                return;
+            }
+            String[] parts = requestLine.split(" ", 3);
+            if (parts.length != 3 || (!"GET".equals(parts[0]) && !"HEAD".equals(parts[0]))) {
+                send(socket, 405, "Method Not Allowed", "text/plain", new byte[0], false);
+                return;
+            }
+            Map<String, String> headers = readHeaders(reader);
+            if (headers == null) {
+                send(socket, 400, "Bad Request", "text/plain", new byte[0], false);
+                return;
+            }
+            String path = parts[1].split("\\?", 2)[0];
+            if ("/health/subscription".equals(path)) {
+                byte[] body = CustomUtil.objectMapper.writeValueAsBytes(snapshotStore.status());
+                send(socket, 200, "OK", "application/json; charset=utf-8", body, "HEAD".equals(parts[0]));
+                return;
+            }
+            String[] segments = path.split("/");
+            if (segments.length != 4 || !"sub".equals(segments[1]) || !("v2".equals(segments[2]) || "cat".equals(segments[2]))) {
+                send(socket, 404, "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            String key = URLDecoder.decode(segments[3], StandardCharsets.UTF_8);
+            if (key.length() < 6 || key.length() > 512) {
+                send(socket, 404, "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            SubscriptionSnapshotStore.Lookup lookup = snapshotStore.lookup(segments[2], key);
+            if (lookup == null) {
+                SubscriptionSnapshotStore.Status status = snapshotStore.status();
+                int code = "unavailable".equals(status.state()) || "expired".equals(status.state()) ? 503 : 404;
+                send(socket, code, code == 503 ? "Service Unavailable" : "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            serveContent(socket, headers.get("range"), lookup.content(), "v2".equals(segments[2]), "HEAD".equals(parts[0]));
+        } catch (Exception e) {
+            log.debug("处理备机订阅请求失败: {}", e.getMessage());
+        }
+    }
+
+    private static Map<String, String> readHeaders(BufferedReader reader) throws IOException {
+        Map<String, String> headers = new HashMap<>();
+        int total = 0;
+        String line;
+        while ((line = reader.readLine()) != null) {
+            total += line.length();
+            if (total > 8192) return null;
+            if (line.isEmpty()) return headers;
+            int colon = line.indexOf(':');
+            if (colon <= 0) return null;
+            headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), line.substring(colon + 1).trim());
+        }
+        return null;
+    }
+
+    private static void serveContent(Socket socket, String range, byte[] content, boolean v2, boolean head) throws IOException {
+        long start = 0;
+        long end = content.length - 1L;
+        int status = 200;
+        String reason = "OK";
+        if (range != null && range.startsWith("bytes=")) {
+            String value = range.substring(6).split(",", 2)[0];
+            String[] values = value.split("-", 2);
+            try {
+                if (values.length != 2 || values[0].isEmpty()) throw new NumberFormatException();
+                start = Long.parseLong(values[0]);
+                if (!values[1].isEmpty()) end = Long.parseLong(values[1]);
+                if (start < 0 || start > end || start >= content.length) throw new NumberFormatException();
+                end = Math.min(end, content.length - 1L);
+                status = 206;
+                reason = "Partial Content";
+            } catch (NumberFormatException e) {
+                send(socket, 416, "Range Not Satisfiable", v2 ? "text/plain" : "text/yaml", new byte[0], head);
+                return;
+            }
+        }
+        byte[] body = Arrays.copyOfRange(content, (int) start, (int) end + 1);
+        send(socket, status, reason, v2 ? "text/plain; charset=utf-8" : "text/yaml; charset=utf-8", body, head,
+                status == 206 ? "bytes " + start + "-" + end + "/" + content.length : null);
+    }
+
+    private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head) throws IOException {
+        send(socket, status, reason, type, body, head, null);
+    }
+
+    private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head, String range) throws IOException {
+        OutputStream output = socket.getOutputStream();
+        StringBuilder header = new StringBuilder()
+                .append("HTTP/1.1 ").append(status).append(' ').append(reason).append("\r\n")
+                .append("Content-Type: ").append(type).append("\r\n")
+                .append("Content-Length: ").append(body.length).append("\r\n")
+                .append("Accept-Ranges: bytes\r\n")
+                .append("Connection: close\r\n");
+        if (range != null) header.append("Content-Range: ").append(range).append("\r\n");
+        header.append("\r\n");
+        output.write(header.toString().getBytes(StandardCharsets.US_ASCII));
+        if (!head) output.write(body);
+        output.flush();
+    }
+}

src/main/java/lion/Main.java

@@ -1,28 +1,28 @@
 package lion;
 
-import lion.ErrorCode.ErrorCode;
+import io.netty.bootstrap.Bootstrap;
-import lombok.extern.log4j.Log4j;
+import lion.Config.Config;
+import lion.Externel.BackupSubServer;
+import lion.Service.SubscriptionSnapshotStore;
+import lombok.extern.slf4j.Slf4j;
 
-import java.io.IOException;
+@Slf4j
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.channels.SocketChannel;
-import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
-
-
-@Log4j
 public class Main {
 
     public static void main(String[] args) {
-
+        boot();
+        Config.loadConfig();
+        SubscriptionSnapshotStore snapshotStore = new SubscriptionSnapshotStore(
+                java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret,
+                Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes);
+        snapshotStore.load();
+        new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers),
+                "subscription-backup-http").start();
         new Thread(() -> MultiThreadedHTTPServer.main(null)).start();
-
+        new storageNode(snapshotStore);
-        new storageNode();
     }
 
-
+    public static void boot(){
-
+        new Bootstrap();
-
+    }
 }

src/main/java/lion/Message/AbstractMessage.java

@@ -10,15 +10,13 @@ public class AbstractMessage {
 
     public static final byte RESPONSE_MESSAGE = 0;
 
-    public static final byte UPDATE_GALLERY_MESSAGE = 4;
-
-    public static final byte GALLERY_PAGE_QUERY_MESSAGE = 5;
-
     public static final byte IDENTITY_MESSAGE = 6;
 
     public static final byte MAINTAIN_MESSAGE = 7;
 
-    public static final byte GALLERY_REQUEST_MESSAGE = 101;
+    public static final byte AVAILABLE_CHECK_MESSAGE = 8;
+
+    public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
 
     public byte messageType;
 

src/main/java/lion/Message/DeleteGalleryMessage.java

@@ -1,20 +0,0 @@
-package lion.Message;
-
-import lombok.Data;
-
-@Data
-public class DeleteGalleryMessage extends AbstractMessage{
-    {
-        messageType = DELETE_GALLERY_MESSAGE;
-    }
-
-    public static final byte DELETE_ALL = 3;
-    public static final byte DELETE_PREVIEW = 2;
-    public static final byte DELETE_SOURCE = 1;
-
-
-
-    byte deleteType;
-
-    String galleryName;
-}

src/main/java/lion/Message/GalleryPageQueryMessage.java

@@ -1,21 +0,0 @@
-package lion.Message;
-
-import com.fasterxml.jackson.annotation.JsonInclude;
-import lombok.Data;
-
-@Data
-public class GalleryPageQueryMessage extends AbstractMessage{
-    {
-        messageType = GALLERY_PAGE_QUERY_MESSAGE;
-    }
-
-    @JsonInclude(JsonInclude.Include.NON_NULL)
-    String name;
-
-    int page;
-
-    @JsonInclude(JsonInclude.Include.NON_NULL)
-    String pageName;
-
-    byte result;
-}

src/main/java/lion/Message/GalleryRequestMessage.java

@@ -1,25 +0,0 @@
-package lion.Message;
-
-import lombok.Data;
-
-@Data
-//请求预览/压缩包
-public class GalleryRequestMessage extends AbstractMessage{
-
-    public static final byte SOURCE = 1;
-    public static final byte PREVIEW = 2;
-
-    public static final byte COMPRESS_SOURCE = 3;
-
-    {
-        messageType = GALLERY_REQUEST_MESSAGE;
-    }
-
-    String galleryName;
-
-    byte type;
-
-    short page;
-
-    short port;
-}

src/main/java/lion/Message/IdentityMessage.java

@@ -1,10 +0,0 @@
-package lion.Message;
-
-import lombok.Data;
-
-@Data
-public class IdentityMessage extends AbstractMessage{
-    {
-        messageType = IDENTITY_MESSAGE;
-    }
-}

src/main/java/lion/Message/Main/AvailableCheckMessage.java

@@ -0,0 +1,11 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+import lombok.Data;
+
+@Data
+public class AvailableCheckMessage extends AbstractMessage {
+    {
+        messageType = AVAILABLE_CHECK_MESSAGE;
+    }
+}

src/main/java/lion/Message/Main/DeleteGalleryMessage.java

@@ -0,0 +1,12 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+import lombok.Data;
+
+@Data
+public class DeleteGalleryMessage extends AbstractMessage {
+    {
+        messageType = DELETE_GALLERY_MESSAGE;
+    }
+    String galleryName;
+}

src/main/java/lion/Message/Main/DownloadPostMessage.java

@@ -1,6 +1,7 @@
-package lion.Message;
+package lion.Message.Main;
 
 import lion.Domain.GalleryTask;
+import lion.Message.AbstractMessage;
 import lombok.Data;
 
 @Data

src/main/java/lion/Message/Main/DownloadStatusMessage.java

@@ -1,6 +1,7 @@
-package lion.Message;
+package lion.Message.Main;
 
 import lion.Domain.GalleryTask;
+import lion.Message.AbstractMessage;
 import lombok.Data;
 
 @Data

src/main/java/lion/Message/Main/IdentityMessage.java

@@ -0,0 +1,12 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+import lombok.Data;
+
+@Data
+public class IdentityMessage extends AbstractMessage {
+    {
+        messageType = IDENTITY_MESSAGE;
+    }
+    String identity;
+}

src/main/java/lion/Message/Main/MaintainMessage.java

@@ -0,0 +1,9 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+
+public class MaintainMessage extends AbstractMessage {
+    {
+        messageType = MAINTAIN_MESSAGE;
+    }
+}

src/main/java/lion/Message/Main/ResponseMessage.java

@@ -1,5 +1,6 @@
-package lion.Message;
+package lion.Message.Main;
 
+import lion.Message.AbstractMessage;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 

src/main/java/lion/Message/Main/SubscriptionAccountSnapshot.java

@@ -0,0 +1,16 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionAccountSnapshot {
+    private Integer accountId;
+    private boolean enabled;
+    private boolean filterHighMultiplier;
+    private String v2ContentBase64;
+    private String v2Sha256;
+    private String clashContentBase64;
+    private String clashSha256;
+}

src/main/java/lion/Message/Main/SubscriptionBindingSnapshot.java

@@ -0,0 +1,11 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionBindingSnapshot {
+    private String publicKeySha256;
+    private Integer accountId;
+}

src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java

@@ -0,0 +1,22 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lion.Message.AbstractMessage;
+
+@Data
+@NoArgsConstructor
+@ToString(exclude = {"payloadBase64", "signature"})
+public class SubscriptionSnapshotMessage extends AbstractMessage {
+    {
+        messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE;
+    }
+
+    private int schemaVersion;
+    private String revision;
+    private long generatedAt;
+    private String payloadBase64;
+    private String payloadSha256;
+    private String signature;
+}

src/main/java/lion/Message/Main/SubscriptionSnapshotPayload.java

@@ -0,0 +1,15 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionSnapshotPayload {
+    private int schemaVersion;
+    private List<SubscriptionAccountSnapshot> accounts = new ArrayList<>();
+    private List<SubscriptionBindingSnapshot> bindings = new ArrayList<>();
+}

src/main/java/lion/Message/MaintainMessage.java

@@ -1,7 +0,0 @@
-package lion.Message;
-
-public class MaintainMessage extends AbstractMessage{
-    {
-        messageType = MAINTAIN_MESSAGE;
-    }
-}

src/main/java/lion/Message/MessageCodec.java

@@ -5,34 +5,39 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageCodec;
-import lombok.extern.log4j.Log4j;
+import lion.CustomUtil;
+import lion.Message.Main.*;
+import lombok.extern.slf4j.Slf4j;
 
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 
-@Log4j
+@Slf4j
 public class MessageCodec extends ByteToMessageCodec<AbstractMessage> {
 
     ObjectMapper objectMapper;
 
     public MessageCodec(){
-        objectMapper = new ObjectMapper();
+        objectMapper = CustomUtil.objectMapper;
     }
 
     @Override
     protected void encode(ChannelHandlerContext channelHandlerContext, AbstractMessage abstractMessage, ByteBuf byteBuf) {
         byteBuf.writeByte(abstractMessage.messageType);
 
-        byte[] bytes = objectMapper.valueToTree(abstractMessage).toString().getBytes(StandardCharsets.UTF_8);
+        try {
+            byte[] bytes = objectMapper.writeValueAsBytes(abstractMessage);
             byteBuf.writeInt(bytes.length);
             byteBuf.writeBytes(bytes);
+        } catch (Exception e) {
+            log.error("序列化消息失败:{}", e.getMessage());
+        }
     }
 
     @Override
     protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
         byte messageType = byteBuf.readByte();
         int length = byteBuf.readInt();
-        System.out.println(length);
         byte[] bytes = new byte[length];
         byteBuf.readBytes(bytes);
         final String metadata = new String(bytes, StandardCharsets.UTF_8);
@@ -40,13 +45,12 @@ public class MessageCodec extends ByteToMessageCodec<AbstractMessage> {
         AbstractMessage abstractMessage = switch (messageType){
             case AbstractMessage.DOWNLOAD_POST_MESSAGE -> objectMapper.readValue(metadata, DownloadPostMessage.class);
             case AbstractMessage.DOWNLOAD_STATUS_MESSAGE -> objectMapper.readValue(metadata, DownloadStatusMessage.class);
-            case AbstractMessage.GALLERY_REQUEST_MESSAGE -> objectMapper.readValue(metadata, GalleryRequestMessage.class);
             case AbstractMessage.RESPONSE_MESSAGE -> objectMapper.readValue(metadata, ResponseMessage.class);
-            case AbstractMessage.UPDATE_GALLERY_MESSAGE -> objectMapper.readValue(metadata, UpdateGalleryMessage.class);
             case AbstractMessage.DELETE_GALLERY_MESSAGE -> objectMapper.readValue(metadata, DeleteGalleryMessage.class);
-            case AbstractMessage.GALLERY_PAGE_QUERY_MESSAGE -> objectMapper.readValue(metadata, GalleryPageQueryMessage.class);
             case AbstractMessage.IDENTITY_MESSAGE -> objectMapper.readValue(metadata, IdentityMessage.class);
             case AbstractMessage.MAINTAIN_MESSAGE -> objectMapper.readValue(metadata, MaintainMessage.class);
+            case AbstractMessage.AVAILABLE_CHECK_MESSAGE -> objectMapper.readValue(metadata, AvailableCheckMessage.class);
+            case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> objectMapper.readValue(metadata, SubscriptionSnapshotMessage.class);
             default -> null;
         };
 

src/main/java/lion/Message/UpdateGalleryMessage.java

@@ -1,13 +0,0 @@
-package lion.Message;
-
-import lion.Domain.GalleryTask;
-import lombok.Data;
-
-@Data
-public class UpdateGalleryMessage extends AbstractMessage{
-    {
-        messageType = UPDATE_GALLERY_MESSAGE;
-    }
-
-    GalleryTask galleryTask;
-}

src/main/java/lion/MultiThreadedHTTPServer.java

@@ -1,5 +1,7 @@
 package lion;
 
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.*;
 import java.net.*;
 import java.nio.charset.StandardCharsets;
@@ -9,49 +11,52 @@ import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
+@Slf4j
 public class MultiThreadedHTTPServer {
     private static final int PORT = 8888;
-    private static final int BUFFER_SIZE = 1024;
 
     public static void main(String[] args) {
         ExecutorService threadPool = Executors.newCachedThreadPool();
         String real_ip;
         try{
             real_ip = InetAddress.getByName("lionwebsite.xyz").getHostAddress();
-        }
+        } catch (UnknownHostException ignored){
-        catch (UnknownHostException ignored){
             real_ip = "207.60.50.74";
         }
         try(ServerSocket serverSocket = new ServerSocket(PORT)) {
-            System.out.println("Server listening on port " + PORT);
+            log.info("Server listening on port {}", PORT);
             while (true) {
                 Socket clientSocket = serverSocket.accept();
                 String ip = clientSocket.getInetAddress().getHostAddress();
                 if(ip.equals(real_ip)){
-                    System.out.println("Client connected");
+                    log.info("Client connected");
-                    // 线程池处理下载请求
                     threadPool.submit(() -> handleClientRequest(clientSocket));
                 }else{
-                    System.out.println("unknown ip: " + ip);
+                    log.info("unknown ip: " + ip);
                     clientSocket.close();
                 }
-
             }
         } catch (IOException e) {
-            e.printStackTrace();
+            log.error("处理http请求时出错,IP:{},ERROR:{}", real_ip, e.getMessage());
         }
     }
 
     private static void handleClientRequest(Socket clientSocket) {
+        String fileName = "";
         try {
             BufferedReader requestReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
             String requestLine = requestReader.readLine();
 
+            if (requestLine == null) {
+                clientSocket.close();
+                return;
+            }
+
             // Parse the request line to get the method and path
             String[] requestParts = requestLine.split(" ");
             String method = requestParts[0];
             Map<String, String> paramMap = parseRequestLine(requestParts[1]);//path
-            System.out.println(Arrays.toString(requestParts));
+            log.info(Arrays.toString(requestParts));
 
             // Only handle GET requests
             if (method.equals("GET")) {
@@ -64,21 +69,27 @@ public class MultiThreadedHTTPServer {
                     }
                     else {
                         String filePath = "/root/gallery/gallery";
+                        String gid = paramMap.get("gid");
+                        file = gid == null ? null : findGalleryZipByGid(new File(filePath), gid);
+
+                        //兼容没有gid参数的旧下载链接，再尝试按链接中的文件名查找
+                        if(file == null){
                             String path = URLDecoder.decode(requestParts[1].split("\\?")[0], StandardCharsets.UTF_8);
-                        if(!path.contains(".")){
+                            if(path.contains(".")){
-                            file = new File("/root/abc");
-                        }else {
                                 String name = path.substring(0, path.lastIndexOf('.'));
-                            System.out.println(name);
+                                name = filePath + name + "/" + name + ".zip";
-                            filePath += (name + "/" + name + ".zip");
+                                file = new File(name);
-                            file = new File(filePath);
+                            }else{
+                                file = new File("/root/abc");
+                            }
                         }
                     }
                 else{
-                    sendErrorResponse(clientSocket, "403 Forbidden");
+                    CustomUtil.sendErrorResponse(clientSocket, "403 Forbidden");
                     return;
                 }
-                System.out.println(file.getAbsolutePath());
+                fileName = file.getName();
+                log.info(file.getAbsolutePath());
                 // Check if the file exists and is readable
                 if (file.exists() && file.isFile() && file.canRead()) {
                     // Get the file length
@@ -87,7 +98,7 @@ public class MultiThreadedHTTPServer {
                     // Get the range information for resuming download
                     long startByte = 0;
                     long endByte = fileLength - 1;
-                    String rangeHeader = getRequestHeader(requestReader, "Range");
+                    String rangeHeader = CustomUtil.getRequestHeader(requestReader);
                     if (rangeHeader != null && rangeHeader.startsWith("bytes=")) {
                         String[] rangeValues = rangeHeader.substring(6).split("-");
                         startByte = Long.parseLong(rangeValues[0]);
@@ -96,59 +107,42 @@ public class MultiThreadedHTTPServer {
                         }
                     }
 
-                    // Send the HTTP response headers
+                    CustomUtil.sendFileRange(clientSocket, file, startByte, endByte, fileName);
-                    OutputStream responseStream = clientSocket.getOutputStream();
-                    PrintWriter responseWriter = new PrintWriter(responseStream, true);
-                    responseWriter.println("HTTP/1.1 206 Partial Content");
-                    responseWriter.println("Content-Type: application/octet-stream");
-                    responseWriter.println("Accept-Ranges: bytes");
-                    responseWriter.println("Content-Length: " + (endByte - startByte + 1));
-                    responseWriter.println("Content-Range: bytes " + startByte + "-" + endByte + "/" + fileLength);
-                    responseWriter.println();
-
-                    // Send the file content
-                    try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
-                        randomAccessFile.seek(startByte);
-                        byte[] buffer = new byte[BUFFER_SIZE];
-                        int bytesRead;
-                        long bytesRemaining = endByte - startByte + 1;
-                        while (bytesRemaining > 0 && (bytesRead = randomAccessFile.read(buffer, 0, (int) Math.min(buffer.length, bytesRemaining))) != -1) {
-                            responseStream.write(buffer, 0, bytesRead);
-                            bytesRemaining -= bytesRead;
-                        }
-                    }catch (SocketException ignore){
-
-                    }
-
-                    // Close the response output stream
-                    responseStream.close();
                 } else {
                     // File not found or not readable, send 404 response
-                    sendErrorResponse(clientSocket, "404 Not Found");
+                    CustomUtil.sendErrorResponse(clientSocket, "404 Not Found");
                 }
             } else {
                 // Non-GET requests, send 501 response
-                sendErrorResponse(clientSocket, "501 Not Implemented");
+                CustomUtil.sendErrorResponse(clientSocket, "501 Not Implemented");
             }
 
             // Close the request reader and client socket
             requestReader.close();
             clientSocket.close();
         } catch (IOException e) {
-            e.printStackTrace();
+            log.error("处理文件下载时出错,IP:{}, 文件:{}, ERROR:{}", clientSocket.getInetAddress().getHostAddress(), fileName, e.getMessage());
         }
     }
 
-    private static String getRequestHeader(BufferedReader requestReader, String headerName) throws IOException {
+    private static File findGalleryZipByGid(File galleryRoot, String gid){
-        String line;
+        File[] galleryDirectories = galleryRoot.listFiles(File::isDirectory);
-        while ((line = requestReader.readLine()) != null) {
+        if(galleryDirectories == null)
-            if (line.trim().isEmpty()) {
+            return null;
-                break;
-            }
 
-            if (line.startsWith(headerName + ":")) {
+        String marker = "[" + gid;
-                return line.substring(headerName.length() + 1).trim();
+        for(File galleryDirectory : galleryDirectories){
-            }
+            String name = galleryDirectory.getName();
+            int markerIndex = name.lastIndexOf(marker);
+            if(markerIndex < 0)
+                continue;
+            int suffixIndex = markerIndex + marker.length();
+            if(suffixIndex >= name.length() || (name.charAt(suffixIndex) != ']' && name.charAt(suffixIndex) != '-'))
+                continue;
+
+            File zip = new File(galleryDirectory, name + ".zip");
+            if(zip.isFile())
+                return zip;
         }
         return null;
     }
@@ -182,14 +176,4 @@ public class MultiThreadedHTTPServer {
         }
         return queryParams;
     }
-
-    private static void sendErrorResponse(Socket clientSocket, String statusCode) throws IOException {
-        OutputStream responseStream = clientSocket.getOutputStream();
-        PrintWriter responseWriter = new PrintWriter(responseStream, true);
-        responseWriter.println("HTTP/1.1 " + statusCode);
-        responseWriter.println("Content-Type: text/html");
-        responseWriter.println();
-        responseWriter.println("<h1>" + statusCode + "</h1>");
-        responseStream.close();
-    }
 }

src/main/java/lion/Service/DeleteService.java

@@ -11,35 +11,9 @@ public class DeleteService {
         if(!file.isDirectory())
             return ErrorCode.FILE_NOT_FOUND;
 
-        if (FileUtil.del(path)) {
+        if (FileUtil.del(path))
             return 0;
-        }else{
+        else
             return ErrorCode.IO_ERROR;
     }
 }
-
-    public static byte deletePreview(String path){
-        File directory = new File(path);
-        File[] files = directory.listFiles((dir, name) -> !name.endsWith("zip"));
-
-        if(files == null)
-            return ErrorCode.FILE_NOT_FOUND;
-        for (File file : files)
-            if(!FileUtil.del(file))
-                return ErrorCode.IO_ERROR;
-
-        return 0;
-    }
-
-    public static byte deleteSource(String path){
-        File directory = new File(path);
-        File[] files = directory.listFiles((dir, name) -> name.endsWith("zip"));
-
-        if(files == null)
-            return ErrorCode.FILE_NOT_FOUND;
-        if(!FileUtil.del(files[0]))
-            return ErrorCode.IO_ERROR;
-
-        return 0;
-    }
-}

src/main/java/lion/Service/DeliveryService.java

@@ -1,102 +0,0 @@
-package lion.Service;
-import lion.ErrorCode.ErrorCode;
-import lion.Message.GalleryPageQueryMessage;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.channels.SocketChannel;
-import java.nio.file.StandardOpenOption;
-import java.util.ArrayList;
-import java.util.Arrays;
-
-import java.util.Comparator;
-import java.util.LinkedHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-public class DeliveryService {
-
-    static String storagePath = "/root/gallery/gallery/";
-    //缓存排序后的页数
-    static LinkedHashMap<String, ArrayList<String>> pageCache;
-
-    static ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
-
-
-    static {
-        pageCache = new LinkedHashMap<>();
-    }
-
-    public static byte deliveryPreview(String name, short page, short port){
-        if(!pageCache.containsKey(name)){
-            byte result;
-            if((result = pageCache(name)) != 0)
-               return result;
-        }
-
-        //0页为缩略图
-        if(page == 0){
-            return delivery(new File(storagePath, name + "/thumbnail.webp"), port);
-        }else{
-            return delivery(new File(storagePath, name + "/" + pageCache.get(name).get(page)), port);
-        }
-    }
-
-    public static byte delivery(File file, short port){
-        if(!file.exists())
-            return ErrorCode.FILE_NOT_FOUND;
-
-        singleThreadPool.submit(() -> {
-            try(SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("lionwebsite.xyz", port));
-                FileChannel fileChannel = FileChannel.open(file.toPath(), StandardOpenOption.READ)) {
-                ByteBuffer buffer = ByteBuffer.allocate(1024);
-                while (fileChannel.read(buffer)!=-1){
-                    buffer.flip();
-                    socketChannel.write(buffer);
-                    buffer.clear();
-                }
-                socketChannel.shutdownOutput();
-                return 0;
-            }catch (IOException e){
-                e.printStackTrace();
-                return ErrorCode.IO_ERROR;
-            }
-        });
-       return 0;
-    }
-
-    public static byte pageQuery(GalleryPageQueryMessage gpqm){
-        if(!pageCache.containsKey(gpqm.getName())){
-            byte result;
-            if((result = pageCache(gpqm.getName())) != 0)
-                return result;
-        }
-        ArrayList<String> pages = pageCache.get(gpqm.getName());
-        if(pages.size() <= gpqm.getPage())
-            return ErrorCode.FILE_NOT_FOUND;
-        gpqm.setPageName(pageCache.get(gpqm.getName()).get(gpqm.getPage()));
-        return 0;
-    }
-
-    public static byte pageCache(String name){
-        File directory = new File(storagePath, name);
-        if(!directory.isDirectory()) {
-            System.out.printf("文件夹%s没找到\n", directory.getAbsolutePath());
-            return ErrorCode.FILE_NOT_FOUND;
-        }
-        ArrayList<String> pageList = new ArrayList<>();
-        File[] files = directory.listFiles(((dir, name1) -> !name1.equals("galleryinfo.txt") && !name1.equals("thumbnail.webp") && !name1.endsWith(".zip")));
-
-        if(files == null)
-            return ErrorCode.FILE_NOT_FOUND;
-        ArrayList<File> fileArrayList = new ArrayList<>(Arrays.asList(files));
-        fileArrayList.sort(Comparator.naturalOrder());
-        pageList.add("thumbnail.webp");
-        fileArrayList.forEach((f) -> pageList.add(f.getName()));
-        pageCache.put(name, pageList);
-        return 0;
-    }
-}

src/main/java/lion/Service/DownloadCheckService.java

@@ -1,28 +1,24 @@
 package lion.Service;
 
+import lion.CustomUtil;
 import lion.Domain.GalleryTask;
-import lion.ErrorCode.ErrorCode;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.util.ZipUtil;
-import lombok.extern.log4j.Log4j;
+import lombok.extern.slf4j.Slf4j;
-import org.im4java.core.ConvertCmd;
-import org.im4java.core.IM4JavaException;
-import org.im4java.core.IMOperation;
 
-import java.io.File;
+import java.io.*;
-import java.io.IOException;
+import java.util.*;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.Iterator;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
+import java.nio.file.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.zip.CRC32;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
 
-
+@Slf4j
-@Log4j
 public class DownloadCheckService {
-    ArrayList<GalleryTask> queue;
+    Map<Integer, GalleryTask> queue;
 
     String downloadPath = "/root/gallery/hentai/download/";
 
@@ -30,19 +26,28 @@ public class DownloadCheckService {
 
     ScheduledThreadPoolExecutor convert_thread;
 
-    ArrayList<GalleryTask> convert_queue;
+    final ArrayList<GalleryTask> compress_queue;
 
-    public DownloadCheckService(ArrayList<GalleryTask> queue){
+    final Map<Integer, Long> retryAfter = new ConcurrentHashMap<>();
+
+    public DownloadCheckService(Map<Integer, GalleryTask> queue){
+        this(queue, true);
+    }
+
+    // Tests use temporary directories and invoke scans explicitly.
+    DownloadCheckService(Map<Integer, GalleryTask> queue, boolean startScheduler){
         this.queue = queue;
-        convert_queue = new ArrayList<>(0);
+        compress_queue = new ArrayList<>(0);
+        if (startScheduler) {
             convert_thread = new ScheduledThreadPoolExecutor(1);
-        convert_thread.scheduleAtFixedRate(this::convert, 0, 5, TimeUnit.SECONDS);
+            convert_thread.scheduleWithFixedDelay(this::compress, 0, 5, TimeUnit.SECONDS);
+        }
     }
 
     public boolean downloadCheck(){
         if(queue.isEmpty())
             return false;
-        log.info("下载检查:" + Arrays.toString(queue.toArray()));
+        log.info("下载检查:{}", Arrays.toString(queue.values().toArray()));
 
         File downloadDirectory = new File(downloadPath);
         File[] fileArray = downloadDirectory.listFiles();
@@ -52,23 +57,35 @@ public class DownloadCheckService {
 
         ArrayList<File> files = new ArrayList<>(Arrays.asList(fileArray));
 
+        boolean result = false;
         //扫描进度
         Iterator<File> fileIterator = files.iterator();
-        for(GalleryTask galleryTask: queue){
+        for(GalleryTask galleryTask: queue.values()){
+            //跳过已经下载完成或者压缩完成的任务
+            if(galleryTask.is_compress_complete() || galleryTask.is_compressing()) {
+                result = true;
+                continue;
+            }
+
             while(fileIterator.hasNext()){
                 File file = fileIterator.next();
-                if(!file.getName().contains(String.valueOf(galleryTask.getGid())))
+                if(!file.isDirectory() || !matchesGid(file.getName(), galleryTask.getGid()))
                     continue;
 
+                galleryTask.setStatus(GalleryTask.DOWNLOADING);
                 fileIterator.remove();
+                if(galleryTask.getName() == null || !galleryTask.getName().equals(file.getName()))
                     galleryTask.setName(file.getName());
 
                 File[] pages = file.listFiles((dir, name) -> !name.equals("galleryinfo.txt"));
                 if (pages == null || pages.length == 0)
                     continue;
+                if(galleryTask.getProceeding() != pages.length)
+                    result = true;
                 galleryTask.setProceeding(pages.length);
 
                 if (new File(file.getPath(), "galleryinfo.txt").exists()) {
+                    result = true;
                     galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
                     galleryTask.setPath(file.getPath());
                 }
@@ -76,81 +93,180 @@ public class DownloadCheckService {
             fileIterator = files.iterator();
         }
 
-        //转格式队列
+        //压缩队列
-        for(GalleryTask galleryTask: queue)
+        for(GalleryTask galleryTask: queue.values())
-            if (galleryTask.is_download_complete())
+            if (galleryTask.is_download_complete()
-                convert_queue.add(galleryTask);
+                    && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) {
+                galleryTask.setStatus(GalleryTask.COMPRESSING);
+                synchronized (compress_queue) {
+                    compress_queue.add(galleryTask);
+                }
+            }
 
-        return true;
+        return result;
     }
 
     /**
-     * 转换线程：将转换队列的任务复制一份，进行转换
+     * 压缩线程：将压缩队列的任务复制一份，进行转换
      */
-    public void convert() {
+    public void compress() {
-        if(convert_queue.isEmpty())
+        ArrayList<GalleryTask> galleryTasks;
+        synchronized (compress_queue) {
+            if (compress_queue.isEmpty())
                 return;
-        ConvertCmd convertCmd = new ConvertCmd(true);
+            galleryTasks = new ArrayList<>(compress_queue);
-        ReentrantLock reentrantLock = new ReentrantLock();
+            compress_queue.clear();
-        reentrantLock.lock();
+        }
-        ArrayList<GalleryTask> galleryTasks = new ArrayList<>(convert_queue);
-        convert_queue.clear();
-        reentrantLock.unlock();
         for (GalleryTask galleryTask : galleryTasks) {
-            File[] images = new File(galleryTask.getPath()).listFiles((dir, name) -> name.endsWith(".jpg") || name.endsWith(".png"));
+            Path temporary = null;
-            if (images == null) {
-                galleryTask.setStatus(ErrorCode.COMPRESS_ERROR);
-                continue;
-            }
-            //长度相同比较字典序，否则比较长度
-            images = Arrays.stream(images).sorted(Comparator.naturalOrder()).toArray(File[]::new);
-
-            //创建文件夹
-            File file = new File(storagePath + galleryTask.getName());
-            if (file.isDirectory() || file.mkdirs()) {
-                log.info(galleryTask.getName() + "文件夹创建成功");
-            } else {
-                log.error(galleryTask.getName() + "文件夹创建失败");
-                continue;
-            }
-
-            //thumbnail
-            IMOperation operation = new IMOperation();
-            operation.addImage(images[0].getAbsolutePath());
-            operation.resize(500, 500);
-            operation.format("webp");
-            operation.addImage(storagePath + galleryTask.getName() + "/thumbnail.webp");
             try {
-                log.info("文件" + images[0].getName() + "，转换为thumbnail.webp");
+                log.info("开始压缩:{}", galleryTask.getName());
-                convertCmd.run(operation);
+                Path directory = Paths.get(storagePath, galleryTask.getName());
-            } catch (IOException | IM4JavaException | InterruptedException e) {
+                Files.createDirectories(directory);
-                log.error("创建" + galleryTask.getName() + "缩略图失败");
+                Path archive = directory.resolve(galleryTask.getName() + ".zip");
-                galleryTask.setStatus(ErrorCode.COMPRESS_ERROR);
+                temporary = Files.createTempFile(directory, ".compress-", ".zip.part");
-                continue;
+                ZipUtil.zip(galleryTask.getPath(), temporary.toString());
-            }
+                if (!isValidArchive(temporary.toFile()))
-
+                    throw new IOException("压缩包校验失败");
-            if ((galleryTask.getType() & GalleryTask.DOWNLOAD_PREVIEW) != 0)
+                // Publish only a closed, verified archive. A crash leaves a .part file.
-                for (int i = 0; i < images.length; i++) {
-                    log.info("文件" + images[i].getName() + "，转换为webp[" + i + "/" + images.length + "]");
-                    operation = new IMOperation();
-                    operation.addImage(images[i].getAbsolutePath());
-                    operation.format("webp");
-                    operation.addImage(storagePath + galleryTask.getName() + "/" + images[i].getName().replace(".png", ".webp").replace(".jpg", ".webp"));
                 try {
-                        convertCmd.run(operation);
+                    Files.move(temporary, archive, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
-                    } catch (IOException | InterruptedException | IM4JavaException e) {
+                } catch (AtomicMoveNotSupportedException e) {
-                        log.error("文件" + images[i].getName() + "转换失败");
+                    Files.move(temporary, archive, StandardCopyOption.REPLACE_EXISTING);
-                        galleryTask.setStatus(ErrorCode.COMPRESS_ERROR);
-                        break;
                 }
-                }
+                temporary = null;
-            if ((galleryTask.getType() & GalleryTask.DOWNLOAD_SOURCE) != 0) {
+                galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE);
-                ZipUtil.zip(galleryTask.getPath(), storagePath + galleryTask.getName() + "/" + galleryTask.getName() + ".zip");
+                retryAfter.remove(galleryTask.getGid());
-                log.info(galleryTask.getName()  + "压缩完成" );
+                if (!FileUtil.del(galleryTask.getPath()))
-            }
+                    log.warn("压缩已完成，但源目录清理失败: {}", galleryTask.getPath());
-            FileUtil.del(galleryTask.getPath());
+                log.info("{}压缩完成", galleryTask.getName());
+            } catch (Exception e) {
+                // Keep the source and restore an existing, retryable protocol state.
+                if (!galleryTask.is_compress_complete()) {
+                    retryAfter.put(galleryTask.getGid(), System.currentTimeMillis() + 30_000);
                     galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
                 }
+                log.error("{}压缩或清理失败，源文件保留，稍后可重试", galleryTask.getName(), e);
+            } finally {
+                if (temporary != null) {
+                    try { Files.deleteIfExists(temporary); }
+                    catch (IOException e) { log.warn("清理压缩临时文件失败: {}", temporary, e); }
+                }
+            }
+        }
     }
 
+    static boolean isValidArchive(File file) {
+        if (!file.isFile())
+            return false;
+        try (ZipFile zip = new ZipFile(file)) {
+            if (zip.size() == 0)
+                return false;
+            byte[] buffer = new byte[8192];
+            Enumeration<? extends ZipEntry> entries = zip.entries();
+            while (entries.hasMoreElements()) {
+                ZipEntry entry = entries.nextElement();
+                if (entry.isDirectory())
+                    continue;
+                CRC32 crc = new CRC32();
+                long size = 0;
+                try (InputStream input = zip.getInputStream(entry)) {
+                    int count;
+                    while ((count = input.read(buffer)) != -1) {
+                        crc.update(buffer, 0, count);
+                        size += count;
+                    }
+                }
+                if (size != entry.getSize() || crc.getValue() != entry.getCrc())
+                    return false;
+            }
+            return true;
+        } catch (IOException e) {
+            return false;
+        }
+    }
+
+    public GalleryTask addToQueue(GalleryTask galleryTask){
+        GalleryTask active = queue.get(galleryTask.getGid());
+        if (active != null && active.is_compressing())
+            return active;
+        retryAfter.remove(galleryTask.getGid());
+        // A reconnect can resend a task whose name is stale (for example, the
+        // downloader appended a resolution suffix). Resolve completed archives
+        // by gid first, because the gid is stable while the directory name is not.
+        File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid());
+        if(storedDirectory != null){
+            queue.remove(galleryTask.getGid());
+            galleryTask.setName(storedDirectory.getName());
+            galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE);
+            CustomUtil.notifyMe(String.format("任务:%s在添加时已下载完成，更新任务状态", galleryTask.getName()));
+            return galleryTask;
+        }
+
+        // If the task is still downloading, retain the actual directory name so
+        // subsequent progress and compression use the same identity.
+        File downloadingDirectory = findDirectoryByGid(new File(downloadPath), galleryTask.getGid());
+        if(downloadingDirectory != null){
+            GalleryTask queuedTask = queue.get(galleryTask.getGid());
+            if(queuedTask == null){
+                refreshDownloadingTask(galleryTask, downloadingDirectory);
+                queue.put(galleryTask.getGid(), galleryTask);
+                return galleryTask;
+            }
+            if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete())
+                refreshDownloadingTask(queuedTask, downloadingDirectory);
+            return queuedTask;
+        }
+
+        GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask);
+        return queuedTask == null ? galleryTask : queuedTask;
+    }
+
+    private void refreshDownloadingTask(GalleryTask galleryTask, File downloadingDirectory){
+        galleryTask.setName(downloadingDirectory.getName());
+        File[] pages = downloadingDirectory.listFiles((dir, name) -> !name.equals("galleryinfo.txt"));
+        galleryTask.setProceeding(pages == null ? 0 : pages.length);
+        if(new File(downloadingDirectory, "galleryinfo.txt").isFile()){
+            galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
+            galleryTask.setPath(downloadingDirectory.getPath());
+        }else{
+            galleryTask.setStatus(GalleryTask.DOWNLOADING);
+        }
+    }
+
+    private File findStoredDirectoryByGid(int gid){
+        File storageDirectory = new File(storagePath);
+        File[] directories = storageDirectory.listFiles(File::isDirectory);
+        if(directories == null)
+            return null;
+
+        for(File directory : directories){
+            if(matchesGid(directory.getName(), gid)
+                    && isValidArchive(new File(directory, directory.getName() + ".zip")))
+                return directory;
+        }
+        return null;
+    }
+
+    private File findDirectoryByGid(File parentDirectory, int gid){
+        File[] directories = parentDirectory.listFiles(File::isDirectory);
+        if(directories == null)
+            return null;
+
+        for(File directory : directories)
+            if(matchesGid(directory.getName(), gid))
+                return directory;
+        return null;
+    }
+
+    private boolean matchesGid(String name, int gid){
+        String gidMarker = "[" + gid;
+        int markerIndex = name.lastIndexOf(gidMarker);
+        if(markerIndex < 0)
+            return false;
+
+        int suffixIndex = markerIndex + gidMarker.length();
+        return suffixIndex < name.length()
+                && (name.charAt(suffixIndex) == ']' || name.charAt(suffixIndex) == '-');
+    }
 }

src/main/java/lion/Service/SubscriptionSnapshotStore.java

@@ -0,0 +1,329 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lion.CustomUtil;
+import lion.Message.Main.SubscriptionAccountSnapshot;
+import lion.Message.Main.SubscriptionBindingSnapshot;
+import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.SubscriptionSnapshotPayload;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.*;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPInputStream;
+
+/**
+ * Last-known-good subscription data used by the standby HTTP server.
+ * The store never downloads upstream subscriptions and never stores upstream keys.
+ */
+@Slf4j
+public final class SubscriptionSnapshotStore {
+    public static final byte APPLY_SUCCESS = 0;
+    public static final byte APPLY_INVALID = 1;
+    public static final byte APPLY_IO_ERROR = 2;
+    public static final byte APPLY_OLD = 3;
+
+    private final Path root;
+    private final byte[] syncSecret;
+    private final long maxStaleMillis;
+    private final int maxPayloadBytes;
+    private final ObjectMapper objectMapper = CustomUtil.objectMapper;
+    private final AtomicReference<Snapshot> current = new AtomicReference<>();
+
+    public SubscriptionSnapshotStore(Path root, String syncSecret, long maxStaleSeconds, int maxPayloadBytes) {
+        this.root = Objects.requireNonNull(root);
+        this.syncSecret = syncSecret == null ? new byte[0] : syncSecret.getBytes(StandardCharsets.UTF_8);
+        this.maxStaleMillis = Math.max(0, maxStaleSeconds) * 1000L;
+        this.maxPayloadBytes = maxPayloadBytes;
+    }
+
+    public void load() {
+        List<String> candidates = new ArrayList<>();
+        try {
+            Path pointer = root.resolve("current-revision");
+            if (Files.isRegularFile(pointer)) {
+                String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim();
+                if (isRevision(revision)) candidates.add(revision);
+            }
+            Path snapshots = root.resolve("snapshots");
+            if (Files.isDirectory(snapshots)) {
+                try (var stream = Files.list(snapshots)) {
+                    stream.filter(Files::isDirectory)
+                            .map(path -> path.getFileName().toString())
+                            .filter(SubscriptionSnapshotStore::isRevision)
+                            .filter(revision -> !candidates.contains(revision))
+                            .sorted(Comparator.reverseOrder())
+                            .forEach(candidates::add);
+                }
+            }
+        } catch (Exception e) {
+            log.warn("扫描订阅快照失败: {}", e.getMessage());
+        }
+        for (String revision : candidates) {
+            try {
+                Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision));
+                current.set(snapshot);
+                writePointer(revision);
+                log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount());
+                return;
+            } catch (Exception e) {
+                log.warn("订阅快照损坏，尝试上一版本 revision={}: {}", shortRevision(revision), e.getMessage());
+            }
+        }
+        current.set(null);
+        log.warn("没有可用的订阅快照，备机订阅暂不可用");
+    }
+
+    public ApplyResult apply(SubscriptionSnapshotMessage message) {
+        Path staging = null;
+        try {
+            if (message == null || message.getSchemaVersion() != 1 || !isRevision(message.getRevision()))
+                return new ApplyResult(APPLY_INVALID, "消息版本或 revision 非法");
+            if (syncSecret.length == 0)
+                return new ApplyResult(APPLY_INVALID, "同步密钥未配置");
+
+            byte[] compressed = decodeBase64(message.getPayloadBase64(), maxPayloadBytes);
+            if (!constantEquals(message.getPayloadSha256(), sha256(compressed)))
+                return new ApplyResult(APPLY_INVALID, "payload SHA-256 校验失败");
+            String signatureInput = message.getSchemaVersion() + "\n" + message.getRevision() + "\n"
+                    + message.getGeneratedAt() + "\n" + message.getPayloadSha256();
+            if (!constantEquals(message.getSignature(), hmac(signatureInput.getBytes(StandardCharsets.UTF_8))))
+                return new ApplyResult(APPLY_INVALID, "快照签名校验失败");
+
+            byte[] payloadBytes = gunzip(compressed, maxPayloadBytes);
+            if (!constantEquals(message.getRevision(), sha256(payloadBytes)))
+                return new ApplyResult(APPLY_INVALID, "revision 与 payload 不一致");
+            SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class);
+            SnapshotData data = validatePayload(payload);
+            Snapshot old = current.get();
+            if (old != null) {
+                if (message.getRevision().equals(old.revision()))
+                    return new ApplyResult(APPLY_OLD, "revision 已存在");
+                if (message.getGeneratedAt() < old.generatedAt())
+                    return new ApplyResult(APPLY_OLD, "快照时间早于当前版本");
+            }
+
+            Path snapshots = root.resolve("snapshots");
+            Files.createDirectories(snapshots);
+            staging = snapshots.resolve(".staging-" + message.getRevision());
+            deleteRecursively(staging);
+            Files.createDirectories(staging.resolve("accounts"));
+            for (AccountFiles account : data.accounts.values()) {
+                Path dir = staging.resolve("accounts").resolve(String.valueOf(account.accountId()));
+                Files.createDirectories(dir);
+                Files.write(dir.resolve("v2ray.txt"), account.v2(), StandardOpenOption.CREATE_NEW);
+                Files.write(dir.resolve("clash.yaml"), account.clash(), StandardOpenOption.CREATE_NEW);
+            }
+            Map<String, Object> manifest = new LinkedHashMap<>();
+            manifest.put("revision", message.getRevision());
+            manifest.put("generatedAt", message.getGeneratedAt());
+            // 保留主站签名对应的原始 JSON 字节，重启时不依赖 Jackson 再序列化顺序。
+            manifest.put("payloadBase64", Base64.getEncoder().encodeToString(payloadBytes));
+            Files.write(staging.resolve("manifest.json"), objectMapper.writeValueAsBytes(manifest), StandardOpenOption.CREATE_NEW);
+
+            Path destination = snapshots.resolve(message.getRevision());
+            if (Files.exists(destination))
+                deleteRecursively(destination);
+            atomicMove(staging, destination);
+            writePointer(message.getRevision());
+
+            Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash,
+                    data.accounts.size(), data.bindingCount);
+            current.set(snapshot);
+            cleanupOldSnapshots(message.getRevision());
+            return new ApplyResult(APPLY_SUCCESS, "同步成功");
+        } catch (Exception e) {
+            log.error("应用订阅快照失败: {}", e.getMessage());
+            return new ApplyResult(APPLY_IO_ERROR, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
+        } finally {
+            if (staging != null && Files.exists(staging)) {
+                try { deleteRecursively(staging); }
+                catch (IOException e) { log.warn("清理订阅快照 staging 失败: {}", staging); }
+            }
+        }
+    }
+
+    private void writePointer(String revision) throws IOException {
+        Files.createDirectories(root);
+        Path pointerTmp = root.resolve("current-revision.tmp");
+        Files.writeString(pointerTmp, revision + "\n", StandardCharsets.UTF_8,
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        atomicMove(pointerTmp, root.resolve("current-revision"));
+    }
+
+    public Lookup lookup(String client, String publicKey) {
+        Snapshot snapshot = current.get();
+        if (snapshot == null || snapshot.expired(System.currentTimeMillis(), maxStaleMillis))
+            return null;
+        AccountFiles account = snapshot.byKeyHash().get(sha256(publicKey.getBytes(StandardCharsets.UTF_8)));
+        if (account == null)
+            return null;
+        return new Lookup("v2".equals(client) ? account.v2() : account.clash(), snapshot.revision(), snapshot.generatedAt());
+    }
+
+    public Status status() {
+        Snapshot snapshot = current.get();
+        if (snapshot == null)
+            return new Status("unavailable", null, 0, 0, 0);
+        long age = Math.max(0, System.currentTimeMillis() - snapshot.generatedAt());
+        boolean expired = snapshot.expired(System.currentTimeMillis(), maxStaleMillis);
+        return new Status(expired ? "expired" : "ready", snapshot.revision(), snapshot.accountCount(), snapshot.bindingCount(), age);
+    }
+
+    private Snapshot loadSnapshot(Path directory) throws IOException {
+        JsonNode manifest = objectMapper.readTree(Files.readAllBytes(directory.resolve("manifest.json")));
+        String revision = manifest.path("revision").asText();
+        long generatedAt = manifest.path("generatedAt").asLong(0);
+        byte[] payloadBytes = decodeBase64(manifest.path("payloadBase64").asText(null), maxPayloadBytes);
+        SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class);
+        SnapshotData data = validatePayload(payload);
+        if (!isRevision(revision) || !constantEquals(revision, sha256(payloadBytes)))
+            throw new IOException("快照 manifest revision 校验失败");
+        Map<Integer, AccountFiles> accounts = new HashMap<>();
+        for (SubscriptionAccountSnapshot account : payload.getAccounts()) {
+            Path dir = directory.resolve("accounts").resolve(String.valueOf(account.getAccountId()));
+            byte[] v2 = Files.readAllBytes(dir.resolve("v2ray.txt"));
+            byte[] clash = Files.readAllBytes(dir.resolve("clash.yaml"));
+            if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash)))
+                throw new IOException("缓存文件校验失败");
+            accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash));
+        }
+        Map<String, AccountFiles> byKey = new HashMap<>();
+        for (SubscriptionBindingSnapshot binding : payload.getBindings())
+            byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId()));
+        return new Snapshot(revision, generatedAt, byKey, accounts.size(), payload.getBindings().size());
+    }
+
+    private SnapshotData validatePayload(SubscriptionSnapshotPayload payload) throws IOException {
+        if (payload == null || payload.getSchemaVersion() != 1 || payload.getAccounts() == null || payload.getBindings() == null)
+            throw new IOException("payload 版本或字段非法");
+        if (payload.getAccounts().size() > 100 || payload.getBindings().size() > 10000)
+            throw new IOException("快照条目数量超限");
+        Map<Integer, AccountFiles> accounts = new HashMap<>();
+        for (SubscriptionAccountSnapshot account : payload.getAccounts()) {
+            if (account == null || account.getAccountId() == null || account.getAccountId() <= 0 || !account.isEnabled()
+                    || !isBase64Sha(account.getV2Sha256()) || !isBase64Sha(account.getClashSha256()))
+                throw new IOException("账号字段非法");
+            byte[] v2 = decodeBase64(account.getV2ContentBase64(), maxPayloadBytes);
+            byte[] clash = decodeBase64(account.getClashContentBase64(), maxPayloadBytes);
+            if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash)))
+                throw new IOException("账号缓存 SHA-256 校验失败");
+            if (accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash)) != null)
+                throw new IOException("账号 ID 重复");
+        }
+        Map<String, AccountFiles> byKey = new HashMap<>();
+        for (SubscriptionBindingSnapshot binding : payload.getBindings()) {
+            if (binding == null || !isBase64Sha(binding.getPublicKeySha256()) || !accounts.containsKey(binding.getAccountId()))
+                throw new IOException("绑定字段或账号引用非法");
+            if (byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId())) != null)
+                throw new IOException("公开 Key Hash 重复");
+        }
+        return new SnapshotData(accounts, byKey, payload.getBindings().size());
+    }
+
+    private void cleanupOldSnapshots(String currentRevision) {
+        try {
+            Path snapshots = root.resolve("snapshots");
+            List<Path> dirs;
+            try (var stream = Files.list(snapshots)) {
+                dirs = stream.filter(Files::isDirectory)
+                        .filter(p -> !p.getFileName().toString().startsWith(".staging-"))
+                        .sorted(Comparator.comparingLong(SubscriptionSnapshotStore::lastModified).reversed())
+                        .toList();
+            }
+            dirs.stream().filter(p -> !p.getFileName().toString().equals(currentRevision))
+                    .skip(1).forEach(p -> {
+                        try { deleteRecursively(p); } catch (IOException e) { log.warn("清理旧订阅快照失败: {}", p); }
+                    });
+        } catch (IOException e) {
+            log.warn("扫描旧订阅快照失败: {}", e.getMessage());
+        }
+    }
+
+    private static byte[] gunzip(byte[] compressed, int limit) throws IOException {
+        try (GZIPInputStream input = new GZIPInputStream(new ByteArrayInputStream(compressed))) {
+            byte[] buffer = new byte[8192];
+            var output = new java.io.ByteArrayOutputStream();
+            int total = 0, read;
+            while ((read = input.read(buffer)) != -1) {
+                total += read;
+                if (total > limit) throw new IOException("解压后 payload 超限");
+                output.write(buffer, 0, read);
+            }
+            return output.toByteArray();
+        }
+    }
+
+    private String hmac(byte[] input) throws Exception {
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(syncSecret, "HmacSHA256"));
+        return hex(mac.doFinal(input));
+    }
+
+    private static byte[] decodeBase64(String value, int maxBytes) throws IOException {
+        if (value == null || value.length() > maxBytes * 2L)
+            throw new IOException("Base64 数据超限");
+        try {
+            byte[] result = Base64.getDecoder().decode(value);
+            if (result.length > maxBytes) throw new IOException("数据超限");
+            return result;
+        } catch (IllegalArgumentException e) {
+            throw new IOException("Base64 数据非法");
+        }
+    }
+
+    private static String sha256(byte[] bytes) {
+        try { return hex(MessageDigest.getInstance("SHA-256").digest(bytes)); }
+        catch (Exception e) { throw new IllegalStateException(e); }
+    }
+
+    private static String hex(byte[] bytes) {
+        return HexFormat.of().formatHex(bytes);
+    }
+
+    private static boolean constantEquals(String expected, String actual) {
+        return expected != null && MessageDigest.isEqual(expected.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII),
+                actual.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII));
+    }
+
+    private static boolean isRevision(String value) { return isBase64Sha(value); }
+    private static boolean isBase64Sha(String value) { return value != null && value.matches("[0-9a-fA-F]{64}"); }
+    private static String shortRevision(String revision) { return revision == null ? null : revision.substring(0, Math.min(12, revision.length())); }
+
+    private static long lastModified(Path path) {
+        try { return Files.getLastModifiedTime(path).toMillis(); }
+        catch (IOException e) { return 0; }
+    }
+
+    private static void atomicMove(Path source, Path target) throws IOException {
+        try { Files.move(source, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); }
+        catch (AtomicMoveNotSupportedException e) { Files.move(source, target, StandardCopyOption.REPLACE_EXISTING); }
+    }
+
+    private static void deleteRecursively(Path path) throws IOException {
+        if (!Files.exists(path)) return;
+        try (var stream = Files.walk(path)) {
+            stream.sorted(Comparator.reverseOrder()).forEach(p -> {
+                try { Files.deleteIfExists(p); } catch (IOException e) { throw new UncheckedIOException(e); }
+            });
+        } catch (UncheckedIOException e) { throw e.getCause(); }
+    }
+
+    public record ApplyResult(byte code, String message) {}
+    public record Lookup(byte[] content, String revision, long generatedAt) {}
+    public record Status(String state, String revision, int accountCount, int bindingCount, long ageMillis) {}
+    private record AccountFiles(Integer accountId, byte[] v2, byte[] clash) {}
+    private record SnapshotData(Map<Integer, AccountFiles> accounts, Map<String, AccountFiles> byKeyHash, int bindingCount) {}
+    private record Snapshot(String revision, long generatedAt, Map<String, AccountFiles> byKeyHash, int accountCount, int bindingCount) {
+        boolean expired(long now, long maxAge) { return maxAge > 0 && now - generatedAt > maxAge; }
+    }
+}

src/main/java/lion/storageNode.java

@@ -1,57 +1,64 @@
 package lion;
 
+import lion.Config.Config;
 import lion.Domain.GalleryTask;
 import lion.Message.*;
+import lion.Message.Main.*;
 import lion.Service.DeleteService;
-import lion.Service.DeliveryService;
+import lion.Service.SubscriptionSnapshotStore;
 import lion.Service.DownloadCheckService;
-import cn.hutool.core.io.FileUtil;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
-import lombok.extern.log4j.Log4j;
+import lombok.extern.slf4j.Slf4j;
 
-
-import java.io.FileOutputStream;
-import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.Socket;
-import java.util.ArrayList;
+import java.util.HashMap;
-import java.util.ListIterator;
+import java.util.Map;
 import java.util.concurrent.*;
 import java.util.concurrent.locks.ReentrantLock;
 
-@Log4j
+@Slf4j
 public class storageNode {
 
-    ChannelFuture channelFuture;
-
     Channel server;
 
+    Channel node;
+
     DownloadCheckService downloadCheckService;
 
-    ArrayList<GalleryTask> queue;
+    Map<Integer, GalleryTask> queue;
 
-    ArrayList<GalleryTask> tempQueue;
+    Map<Integer, GalleryTask> tempQueue;
 
     ScheduledExecutorService checkThreadPool;
 
-    int counter;
-
     ReentrantLock lock;
 
+    final SubscriptionSnapshotStore subscriptionSnapshotStore;
+
+    final ExecutorService subscriptionApplyExecutor;
+
     public static String storagePath = "/root/gallery/gallery/";
 
-    public storageNode(){
+    public storageNode(SubscriptionSnapshotStore subscriptionSnapshotStore){
-        queue = new ArrayList<>(0);
+        this.subscriptionSnapshotStore = subscriptionSnapshotStore;
-        tempQueue = new ArrayList<>(0);
+        this.subscriptionApplyExecutor = Executors.newSingleThreadExecutor(r -> {
+            Thread thread = new Thread(r, "subscription-snapshot-apply");
+            thread.setDaemon(true);
+            return thread;
+        });
+        queue = new ConcurrentHashMap<>();
+        tempQueue = new HashMap<>();
         lock = new ReentrantLock();
-        counter = 0;
 
-        channelFuture = new ServerBootstrap()
+        int real_port = CustomUtil._findIdlePort(26321);
+
+        new ServerBootstrap()
                 .channel(NioServerSocketChannel.class)
                 .group(new NioEventLoopGroup())
                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
@@ -60,16 +67,26 @@ public class storageNode {
                         channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(100000000, 1, 4));
                         channel.pipeline().addLast(new MessageCodec());
                         channel.pipeline().addLast(new MyChannelInboundHandlerAdapter(tempQueue));
-
                     }
-                }).option(ChannelOption.SO_BACKLOG, 128)
+                }).bind(real_port);
-                .childOption(ChannelOption.SO_KEEPALIVE, true)
+        log.info("listening on port {}", real_port);
-                .bind(26321);
-        log.info("listen port:8080");
 
+        int i;
+        for(i=0; i<=20; i++) {
             try (Socket socket = new Socket()) {
-            socket.connect(new InetSocketAddress("lionwebsite.xyz", 26322));
+                socket.setSoTimeout(3000);
-        } catch (Exception ignored) {}
+                log.info("wake up main server on port {}", 26322 + i);
+                socket.connect(new InetSocketAddress("lionwebsite.xyz", 26322 + i));
+                byte[] bytes = socket.getInputStream().readAllBytes();
+                if(bytes.length > 0 && new String(bytes).equals("lionwebsite")) {
+                    break;
+                }
+            } catch (Exception ignored) {
+            }
+        }
+        if (i==20) {
+            log.info("server connect failed");
+        }
         downloadCheckService = new DownloadCheckService(queue);
         checkThreadPool = Executors.newScheduledThreadPool(1);
         checkThreadPool.scheduleAtFixedRate(this::mainThread, 5, 5, TimeUnit.SECONDS);
@@ -78,13 +95,25 @@ public class storageNode {
     public void mainThread(){
         try {
             lock.lock();
+            try {
                 if(!tempQueue.isEmpty()){
-                queue.addAll(tempQueue);
+                    queue.putAll(tempQueue);
                     tempQueue.clear();
                 }
+            } finally {
                 lock.unlock();
-            //检查
+            }
+            //检查，当任务状态发生变化即方法返回true时，再更新，否则return
             if (!downloadCheckService.downloadCheck()) {
+                boolean isSkip = true;
+                //返回false之后，还要额外检查是否有压缩完成的任务
+                if(!queue.isEmpty())
+                    for (GalleryTask galleryTask : queue.values())
+                        if (galleryTask.is_compress_complete()) {
+                            isSkip = false;
+                            break;
+                        }
+                if(isSkip) {
                     counter++;
                     if (server != null && server.isActive() && counter > 10) {
                         server.writeAndFlush(new MaintainMessage());
@@ -92,96 +121,105 @@ public class storageNode {
                     }
                     return;
                 }
+            }
             //发送
+            //上锁后再发送，避免出现发送完之后再下载完成
+            lock.lock();
+            try {
                 DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
-            downloadStatusMessage.setGalleryTasks(queue.toArray(GalleryTask[]::new));
+                downloadStatusMessage.setGalleryTasks(queue.values().toArray(GalleryTask[]::new));
                 server.writeAndFlush(downloadStatusMessage);
 
-            ListIterator<GalleryTask> listIterator = queue.listIterator();
+                queue.entrySet().removeIf(entry -> entry.getValue().is_compress_complete());
-            lock.lock();
-            while (listIterator.hasNext()) {
-                GalleryTask galleryTask = listIterator.next();
-                if (galleryTask.is_download_complete())
-                    listIterator.remove();
-            }
-            lock.unlock();
                 log.info("任务状态发送完成");
+            } finally {
+                lock.unlock();
+            }
         }catch (Exception e){
-            e.printStackTrace();
+            log.error("发送任务状态时发生异常:{}", e.getMessage());
-            try (OutputStream outputStream = new FileOutputStream("/root/gallery/storageNode/err.txt")){
-                outputStream.write(e.getMessage().getBytes());
-                channelFuture.channel().close().sync();
-                System.exit(-1);
-            }catch (Exception ex){
-                ex.printStackTrace();
-            }
         }
     }
 
+    int counter;
+
     class MyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
-        ArrayList<GalleryTask> queue;
+        Map<Integer, GalleryTask> queue;
 
-        public MyChannelInboundHandlerAdapter(ArrayList<GalleryTask> queue) {
+        public MyChannelInboundHandlerAdapter(Map<Integer, GalleryTask> queue) {
-            super();
             this.queue = queue;
         }
 
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) {
-            System.out.println(msg);
+            if (msg instanceof SubscriptionSnapshotMessage snapshot)
+                log.info("收到订阅快照 revision={}", shortRevision(snapshot.getRevision()));
+            else
+                log.info(String.valueOf(msg));
+            AbstractMessage abstractMessage = (AbstractMessage) msg;
 
-            if(msg instanceof IdentityMessage) {
+            switch (abstractMessage.messageType){
+                case AbstractMessage.IDENTITY_MESSAGE -> {
+                    IdentityMessage identityMessage = (IdentityMessage) abstractMessage;
+                    if(identityMessage.getIdentity().equals("lionwebsite")) {
                         server = ctx.channel();
                         log.info("server 上线");
-                //提交下载
+                    } else if(identityMessage.getIdentity().equals("lionwebsiteside")){
-            }else if(msg instanceof DownloadPostMessage dpm){
+                        node = ctx.channel();
+                        log.info("node上线");
+                    }
+                }
+                case AbstractMessage.DOWNLOAD_POST_MESSAGE -> {
+                    DownloadPostMessage dpm = (DownloadPostMessage) abstractMessage;
                     lock.lock();
-                queue.add(dpm.getGalleryTask());
+                    try {
-                System.out.println(queue);
+                        //每次收到任务都重新检查并立即回传当前状态，供主站的单任务重试接口使用
+                        GalleryTask currentTask = downloadCheckService.addToQueue(dpm.getGalleryTask());
+                        DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
+                        downloadStatusMessage.setGalleryTasks(new GalleryTask[]{currentTask});
+                        server.writeAndFlush(downloadStatusMessage);
+                        log.info(String.valueOf(queue));
+                    } finally {
                         lock.unlock();
+                    }
                     ctx.writeAndFlush(new ResponseMessage(dpm.messageId, (byte) 0));
-
+                }
-            //删除本子 全部/预览/源文件
+                case AbstractMessage.DELETE_GALLERY_MESSAGE -> {
-            }else if(msg instanceof DeleteGalleryMessage deleteGalleryMessage){
+                    DeleteGalleryMessage deleteGalleryMessage = (DeleteGalleryMessage) abstractMessage;
-               byte result =  switch (deleteGalleryMessage.getDeleteType()){
+                    String galleryName = deleteGalleryMessage.getGalleryName();
-                    case DeleteGalleryMessage.DELETE_ALL ->
+                    lock.lock();
-                        DeleteService.deleteAll(storagePath + deleteGalleryMessage.getGalleryName());
+                    try {
-                    case DeleteGalleryMessage.DELETE_PREVIEW ->
+                        // 删除任务时也要从待下载队列移除，避免继续向主站上报已经删除的任务状态。
-                        DeleteService.deletePreview(storagePath + deleteGalleryMessage.getGalleryName());
+                        storageNode.this.queue.entrySet().removeIf(entry -> galleryName.equals(entry.getValue().getName()));
-                    case DeleteGalleryMessage.DELETE_SOURCE ->
+                        queue.entrySet().removeIf(entry -> galleryName.equals(entry.getValue().getName()));
-                        DeleteService.deleteSource(storagePath + deleteGalleryMessage.getGalleryName());
+                    } finally {
-                    default -> -1;
+                        lock.unlock();
-                };
+                    }
+                    byte result = DeleteService.deleteAll(storagePath + galleryName);
                     ResponseMessage responseMessage = new ResponseMessage(deleteGalleryMessage.messageId, result);
                     ctx.writeAndFlush(responseMessage);
-
-            //请求预览
-            }else if(msg instanceof GalleryRequestMessage grm){
-                byte result = DeliveryService.deliveryPreview(grm.getGalleryName(), grm.getPage(), grm.getPort());
-
-                ResponseMessage responseMessage = new ResponseMessage();
-                responseMessage.messageId = grm.messageId;
-                responseMessage.setResult(result);
-                ctx.writeAndFlush(responseMessage);
-
-            //更新本子  删除原有文件，然后加入到队列
-            }else if(msg instanceof UpdateGalleryMessage ugm){
-                FileUtil.del(storagePath + ugm.getGalleryTask().getName());
-                lock.lock();
-                queue.add(ugm.getGalleryTask());
-                System.out.println(queue);
-                lock.unlock();
-                ctx.writeAndFlush(new ResponseMessage(ugm.messageId, (byte) 0));
-
-            }else if(msg instanceof GalleryPageQueryMessage gpqm){
-                byte result = DeliveryService.pageQuery(gpqm);
-                gpqm.setResult(result);
-                ctx.writeAndFlush(gpqm);
                 }
-
+                case AbstractMessage.AVAILABLE_CHECK_MESSAGE -> {
-            //修复预览
+                    AvailableCheckMessage acm = (AvailableCheckMessage) abstractMessage;
-
+                    ResponseMessage responseMessage = new ResponseMessage(acm.messageId, (byte)0);
-            //重新生成压缩包
+                    ctx.writeAndFlush(responseMessage);
+                }
+                case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> {
+                    SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage;
+                    if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(server)) {
+                        ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID));
+                        return;
+                    }
+                    subscriptionApplyExecutor.execute(() -> {
+                        SubscriptionSnapshotStore.ApplyResult result = subscriptionSnapshotStore.apply(snapshotMessage);
+                        ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, result.code()));
+                        log.info("订阅快照处理完成 revision={} result={}", shortRevision(snapshotMessage.getRevision()), result.code());
+                    });
+                }
+            }
+//
+//            //修复预览
+//
+//            //重新生成压缩包
         }
 
         @Override
@@ -189,8 +227,14 @@ public class storageNode {
             if(ctx.channel().equals(server)) {
                 log.info("server 下线");
                 server = null;
-            }
+            } else if(ctx.channel().equals(node)){
+                log.info("node 下线");
+                node = null;
             }
         }
     }
 
+    private static String shortRevision(String revision) {
+        return revision == null ? null : revision.substring(0, Math.min(12, revision.length()));
+    }
+}

src/main/resources/config.properties

@@ -0,0 +1,9 @@
+# 订阅快照由主站通过 Netty 长连接推送；不要在此保存上游订阅地址或 Key。
+SubscriptionSyncEnabled=false
+SubscriptionSyncSecret=
+SubscriptionDataDir=/root/gallery/storageNode/sub
+SubscriptionMaxStaleSeconds=604800
+SubscriptionMaxPayloadBytes=52428800
+SubscriptionHttpPort=8889
+SubscriptionHttpWorkers=4
+SubscriptionSocketTimeoutMs=10000

src/main/resources/logback.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>run.out</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+            <fileNamePattern>run.%i.out</fileNamePattern>
+            <minIndex>1</minIndex>
+            <maxIndex>10</maxIndex>
+        </rollingPolicy>
+        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+            <maxFileSize>10MB</maxFileSize>
+        </triggeringPolicy>
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%level] %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="info">
+        <appender-ref ref="FILE"/>
+    </root>
+</configuration>

src/main/resources/reflect-config.json

@@ -0,0 +1,128 @@
+[
+  {
+    "name": "lion.Message.Main.DeleteGalleryMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.DownloadPostMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.DownloadStatusMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.IdentityMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.MaintainMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.ResponseMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.AvailableCheckMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionSnapshotMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionSnapshotPayload",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionAccountSnapshot",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionBindingSnapshot",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Service.SubscriptionSnapshotStore$Status",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.AbstractMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Domain.GalleryTask",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  }
+]

src/test/java/lion/Service/DownloadCheckServiceTest.java

@@ -0,0 +1,51 @@
+package lion.Service;
+
+import lion.Domain.GalleryTask;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.*;
+import java.util.concurrent.ConcurrentHashMap;
+import static org.junit.jupiter.api.Assertions.*;
+
+class DownloadCheckServiceTest {
+    @Test
+    void failedCompressionRetainsSourceAndCanRetry(@TempDir Path root) throws Exception {
+        var queue = new ConcurrentHashMap<Integer, GalleryTask>();
+        var service = new DownloadCheckService(queue, false);
+        Path downloads = Files.createDirectory(root.resolve("downloads"));
+        Path source = Files.createDirectory(downloads.resolve("sample [123]"));
+        Files.writeString(source.resolve("galleryinfo.txt"), "metadata");
+        Files.writeString(source.resolve("1.jpg"), "image bytes");
+        Path storage = root.resolve("storage");
+        Files.writeString(storage, "block mkdir");
+        service.downloadPath = downloads.toString();
+        service.storagePath = storage.toString();
+        GalleryTask task = new GalleryTask();
+        task.setGid(123);
+        service.addToQueue(task);
+        service.downloadCheck();
+        service.compress();
+        assertEquals(GalleryTask.DOWNLOAD_COMPLETE, task.getStatus());
+        assertTrue(Files.exists(source.resolve("1.jpg")));
+        Files.delete(storage);
+        Files.createDirectory(storage);
+        service.addToQueue(task); // manual retry removes the backoff
+        service.downloadCheck();
+        service.compress();
+        assertEquals(GalleryTask.COMPRESS_COMPLETE, task.getStatus());
+        assertTrue(DownloadCheckService.isValidArchive(storage.resolve("sample [123]/sample [123].zip").toFile()));
+        assertFalse(Files.exists(source));
+    }
+
+    @Test
+    void corruptArchiveIsNotTreatedAsCompleted(@TempDir Path root) throws Exception {
+        Path stored = Files.createDirectories(root.resolve("stored/sample [123]"));
+        Files.writeString(stored.resolve("sample [123].zip"), "partial zip");
+        var service = new DownloadCheckService(new ConcurrentHashMap<>(), false);
+        service.storagePath = root.resolve("stored").toString();
+        service.downloadPath = root.resolve("downloads").toString();
+        GalleryTask task = new GalleryTask();
+        task.setGid(123);
+        assertFalse(service.addToQueue(task).is_compress_complete());
+    }
+}

src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java

@@ -0,0 +1,104 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lion.CustomUtil;
+import lion.Message.Main.SubscriptionAccountSnapshot;
+import lion.Message.Main.SubscriptionBindingSnapshot;
+import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.SubscriptionSnapshotPayload;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.Base64;
+import java.util.HexFormat;
+import java.util.zip.GZIPOutputStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SubscriptionSnapshotStoreTest {
+    private static final String SECRET = "snapshot-test-secret";
+    private final ObjectMapper mapper = CustomUtil.objectMapper;
+
+    @Test
+    void appliesSnapshotAndServesByHashedPublicKey(@TempDir Path directory) throws Exception {
+        SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
+
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+        SubscriptionSnapshotStore.Lookup v2 = store.lookup("v2", "public-key-1");
+        SubscriptionSnapshotStore.Lookup clash = store.lookup("cat", "public-key-1");
+        assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), v2.content());
+        assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), clash.content());
+        assertEquals(SubscriptionSnapshotStore.APPLY_OLD, store.apply(message).code());
+    }
+
+    @Test
+    void rejectsTamperedPayloadAndKeepsPreviousSnapshot(@TempDir Path directory) throws Exception {
+        SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+        message.setPayloadBase64(Base64.getEncoder().encodeToString("tampered".getBytes(StandardCharsets.UTF_8)));
+        assertEquals(SubscriptionSnapshotStore.APPLY_INVALID, store.apply(message).code());
+        assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), store.lookup("v2", "public-key-1").content());
+    }
+
+    @Test
+    void loadsLastGoodSnapshotAfterRestart(@TempDir Path directory) throws Exception {
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
+        SubscriptionSnapshotStore first = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, first.apply(message).code());
+        SubscriptionSnapshotStore second = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        second.load();
+        assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), second.lookup("cat", "public-key-1").content());
+    }
+
+    private SubscriptionSnapshotMessage message(String publicKey, String v2, String clash, long generatedAt) throws Exception {
+        byte[] v2Bytes = v2.getBytes(StandardCharsets.UTF_8);
+        byte[] clashBytes = clash.getBytes(StandardCharsets.UTF_8);
+        SubscriptionAccountSnapshot account = new SubscriptionAccountSnapshot();
+        account.setAccountId(1);
+        account.setEnabled(true);
+        account.setV2ContentBase64(Base64.getEncoder().encodeToString(v2Bytes));
+        account.setV2Sha256(sha256(v2Bytes));
+        account.setClashContentBase64(Base64.getEncoder().encodeToString(clashBytes));
+        account.setClashSha256(sha256(clashBytes));
+        SubscriptionBindingSnapshot binding = new SubscriptionBindingSnapshot();
+        binding.setPublicKeySha256(sha256(publicKey.getBytes(StandardCharsets.UTF_8)));
+        binding.setAccountId(1);
+        SubscriptionSnapshotPayload payload = new SubscriptionSnapshotPayload();
+        payload.setSchemaVersion(1);
+        payload.setAccounts(java.util.List.of(account));
+        payload.setBindings(java.util.List.of(binding));
+        byte[] json = mapper.writeValueAsBytes(payload);
+        byte[] compressed = gzip(json);
+        SubscriptionSnapshotMessage message = new SubscriptionSnapshotMessage();
+        message.setSchemaVersion(1);
+        message.setRevision(sha256(json));
+        message.setGeneratedAt(generatedAt);
+        message.setPayloadBase64(Base64.getEncoder().encodeToString(compressed));
+        message.setPayloadSha256(sha256(compressed));
+        String input = "1\n" + message.getRevision() + "\n" + generatedAt + "\n" + message.getPayloadSha256();
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+        message.setSignature(hex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8))));
+        return message;
+    }
+
+    private static byte[] gzip(byte[] bytes) throws Exception {
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        try (GZIPOutputStream gzip = new GZIPOutputStream(output)) { gzip.write(bytes); }
+        return output.toByteArray();
+    }
+
+    private static String sha256(byte[] bytes) throws Exception {
+        return hex(MessageDigest.getInstance("SHA-256").digest(bytes));
+    }
+
+    private static String hex(byte[] bytes) { return HexFormat.of().formatHex(bytes); }
+}