支持订阅快照备机同步与按用户分发

pom.xml

@@ -64,10 +64,22 @@
             <artifactId>httpclient</artifactId>
             <version>4.5.14</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.10.2</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>3.2.5</version>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
@@ -128,4 +140,4 @@
         </plugins>
     </build>
 
-</project>
+</project>

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

@@ -9,18 +9,37 @@ import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class Config {
-    public static String DouNaiV2ray;
-    public static String DouNaiClash;
+    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);
-            DouNaiV2ray = prop.getProperty("DouNaiV2ray");
-            DouNaiClash = prop.getProperty("DouNaiClash");
+            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/Externel/BackupSubServer.java

@@ -1,235 +1,153 @@
 package lion.Externel;
 
+import lion.Config.Config;
+import lion.CustomUtil;
+import lion.Service.SubscriptionSnapshotStore;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpEntity;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
 
 import java.io.*;
-import java.net.*;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.util.*;
-import java.util.concurrent.*;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import lion.CustomUtil;
-import static lion.Config.Config.DouNaiClash;
-import static lion.Config.Config.DouNaiV2ray;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
+/** HTTP distributor for the last-known-good subscription snapshot. */
 @Slf4j
-public class BackupSubServer {
+public final class BackupSubServer implements Runnable {
+    private final SubscriptionSnapshotStore snapshotStore;
+    private final int port;
+    private final ExecutorService workers;
 
-    public static void main(String[] args) {
-        updateSub();
-        ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
-        threadPool.scheduleAtFixedRate(BackupSubServer::updateSub, 0, 12, TimeUnit.HOURS);
+    public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) {
+        this.snapshotStore = Objects.requireNonNull(snapshotStore);
+        this.port = port;
+        this.workers = Executors.newFixedThreadPool(Math.max(1, workerCount));
+    }
 
-        String ip = "";
-        try(ServerSocket serverSocket = new ServerSocket(8889)) {
-            log.info("Sub Server listening on port {}", 8889);
-            while (true) {
-                Socket clientSocket = serverSocket.accept();
-                ip = clientSocket.getInetAddress().getHostAddress();
-                log.info("Client connected:{}", ip);
-                handleClientRequest(clientSocket);
+    @Override
+    public void run() {
+        snapshotStore.load();
+        try (ServerSocket serverSocket = new ServerSocket(port)) {
+            log.info("备机订阅服务监听端口 {}", port);
+            while (!Thread.currentThread().isInterrupted()) {
+                Socket socket = serverSocket.accept();
+                workers.execute(() -> handle(socket));
             }
         } catch (IOException e) {
-            log.error("处理http请求时出错,IP:{},ERROR:{}", ip, e.getMessage());
+            log.error("备机订阅服务停止: {}", e.getMessage());
+        } finally {
+            workers.shutdownNow();
         }
     }
 
-    public static void updateSub(){
-        File DouNaiClashFile = new File("sub/DouNaiClash.txt");
-        File DouNaiV2rayFile = new File("sub/DouNaiV2ray.txt");
-        File directory = new File("sub");
-
-        if(!directory.isDirectory())
-            try {
-                Files.createDirectory(Paths.get("sub"));
-            } catch (IOException e) {
-                log.error("create directory error:{}", e.getMessage());
-            }
-
-        List<String> DouNaiClash_profile;
-
-        //下载豆奶v2ray订阅
-        try(FileWriter writer = new FileWriter(DouNaiV2rayFile)) {
-            String DouNaiV2rayRaw = Get(DouNaiV2ray).getFirst();
-            String[] v2rayPlain = new String(Base64.getDecoder().decode(DouNaiV2rayRaw)).split("\n");
-            StringBuilder stringBuilder = new StringBuilder();
-            Pattern pattern = Pattern.compile("-?\\d+(\\.\\d+)?");
-
-            //过滤高倍率节点
-            for(String node: v2rayPlain){
-                String name = URLDecoder.decode(node.split("#")[1], StandardCharsets.UTF_8);
-                if(name.contains("流量")){
-                    Matcher matcher = pattern.matcher(name.substring(name.indexOf("(") + 1, name.indexOf(")")));
-
-                    if (matcher.find()) {
-                        float ratio = Float.parseFloat(matcher.group());
-                        if(ratio <= 2) {
-                            stringBuilder.append(node).append("\n");
-                            continue;
-                        }
-                    }
-                    stringBuilder.append(node).append("\n");
-                }
-                else{
-                    stringBuilder.append(node).append("\n");
-                }
-            }
-            writer.write(new String(Base64.getEncoder().encode(stringBuilder.toString().getBytes(StandardCharsets.UTF_8))));
-
-            log.info("load DouNai v2ray complete");
-        }catch (IOException e){
-            log.error("load DouNai v2ray failure: {}", e.getMessage());
-        }
-
-        //下载豆奶clash订阅
-        try(FileWriter writer = new FileWriter(DouNaiClashFile)) {
-            DouNaiClash_profile = Get(DouNaiClash);
-            //过滤高倍率节点
-            ArrayList<String> clashProcessed = new ArrayList<>();
-            boolean isProxies = false;
-            boolean skip = false;
-            for(String line: DouNaiClash_profile){
-                if(line.equals("proxies:"))
-                    isProxies = true;
-                else if(line.equals("proxy-groups:") && isProxies)
-                    isProxies = false;
-
-                if(isProxies) {
-                    if (line.contains("name"))
-                        skip = line.contains("流量");
-                    if (!skip)
-                        clashProcessed.add(line);
-                }
-                else
-                if (!line.contains("流量"))
-                    clashProcessed.add(line);
-            }
-
-            for(String line: clashProcessed)
-                writer.write(line + "\n");
-
-            log.info("load DouNai clash complete");
-        }catch (IOException e){
-            log.error("load DouNai clash failure: {}", e.getMessage());
-        }
-    }
-
-
-    public static ArrayList<String> Get(String url) throws IOException {
-        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
-            HttpGet httpGet = new HttpGet(url);
-            try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
-                HttpEntity responseEntity = httpResponse.getEntity();
-                int statusCode = httpResponse.getStatusLine().getStatusCode();
-                ArrayList<String> temp = new ArrayList<>();
-
-                if (statusCode == 200) {
-                    BufferedReader reader = new BufferedReader(new InputStreamReader(responseEntity.getContent()));
-                    String str;
-                    while ((str = reader.readLine()) != null)
-                        temp.add(str);
-                }
-                return temp;
-            }
-        }
-    }
-
-    private static void handleClientRequest(Socket clientSocket) {
-        String fileName = "";
-        try {
-            BufferedReader requestReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
-            String requestLine = requestReader.readLine();
-
-            // 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
-            if(paramMap == null){
-                CustomUtil.sendErrorResponse(clientSocket, "404");
+    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;
             }
-            log.info(Arrays.toString(requestParts));
-
-            // Only handle GET requests
-            if (method.equals("GET")) {
-                // Set the file path for download
-                File file = new File(fileName);
-                switch (paramMap.get("Client")) {
-                    case "v2" -> file = new File("sub/DouNaiV2ray.txt");
-                    case "cat" -> file = new File("sub/DouNaiClash.txt");
-                }
-                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
-                    long fileLength = file.length();
-
-                    // Get the range information for resuming download
-                    long startByte = 0;
-                    long endByte = fileLength - 1;
-                    String rangeHeader = CustomUtil.getRequestHeader(requestReader);
-                    if (rangeHeader != null && rangeHeader.startsWith("bytes=")) {
-                        String[] rangeValues = rangeHeader.substring(6).split("-");
-                        startByte = Long.parseLong(rangeValues[0]);
-                        if (rangeValues.length > 1 && !rangeValues[1].isEmpty()) {
-                            endByte = Long.parseLong(rangeValues[1]);
-                        }
-                    }
-
-                    CustomUtil.sendFileRange(clientSocket, file, startByte, endByte);
-                } else {
-                    // File not found or not readable, send 404 response
-                    CustomUtil.sendErrorResponse(clientSocket, "404 Not Found");
-                }
-            } else {
-                // Non-GET requests, send 501 response
-                CustomUtil.sendErrorResponse(clientSocket, "501 Not Implemented");
+            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;
             }
-
-            // Close the request reader and client socket
-            requestReader.close();
-            clientSocket.close();
-        } catch (IOException e) {
-            log.error("处理文件下载时出错,IP:{}, 文件:{}, ERROR:{}", clientSocket.getInetAddress().getHostAddress(), fileName, e.getMessage());
+            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());
         }
     }
 
-    public static Map<String, String> parseRequestLine(String requestLine) {
-        Map<String, String> pathParams = new HashMap<>();
-
-        if(requestLine == null)
-            return null;
-
-        String path;
-        if(requestLine.contains("?"))
-            path = requestLine.split("\\?")[0];
-        else
-            path = requestLine;
-
-        String[] vars = path.split("/");
-
-        if(vars.length < 4)
-            return null;
-
-        pathParams.put("Key", vars[3]);
-        pathParams.put("Client", vars[2]);
-
-        if(!pathParams.get("Client").equals("cat") && !pathParams.get("Client").equals("v2"))
-            return null;
-
-        if(pathParams.get("Key").length()<6)
-            return null;
-
-        return pathParams;
+    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

@@ -3,6 +3,7 @@ package lion;
 import io.netty.bootstrap.Bootstrap;
 import lion.Config.Config;
 import lion.Externel.BackupSubServer;
+import lion.Service.SubscriptionSnapshotStore;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -11,9 +12,13 @@ public class Main {
     public static void main(String[] args) {
         boot();
         Config.loadConfig();
-        new Thread(() -> BackupSubServer.main(null)).start();
+        SubscriptionSnapshotStore snapshotStore = new SubscriptionSnapshotStore(
+                java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret,
+                Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes);
+        new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers),
+                "subscription-backup-http").start();
         new Thread(() -> MultiThreadedHTTPServer.main(null)).start();
-        new storageNode();
+        new storageNode(snapshotStore);
     }
 
     public static void boot(){

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

@@ -16,6 +16,8 @@ public class AbstractMessage {
 
     public static final byte AVAILABLE_CHECK_MESSAGE = 8;
 
+    public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
+
     public byte messageType;
 
     public int messageId;

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,20 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lion.Message.AbstractMessage;
+
+@Data
+@NoArgsConstructor
+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/MessageCodec.java

@@ -50,6 +50,7 @@ public class MessageCodec extends ByteToMessageCodec<AbstractMessage> {
             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/Service/SubscriptionSnapshotStore.java

@@ -0,0 +1,289 @@
+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.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() {
+        try {
+            Path pointer = root.resolve("current-revision");
+            if (!Files.isRegularFile(pointer)) {
+                log.warn("没有找到订阅快照，备机订阅暂不可用");
+                return;
+            }
+            String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim();
+            if (!isRevision(revision))
+                throw new IOException("当前快照 revision 非法");
+            Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision));
+            current.set(snapshot);
+            log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount());
+        } catch (Exception e) {
+            current.set(null);
+            log.error("加载订阅快照失败，备机订阅暂不可用: {}", e.getMessage());
+        }
+    }
+
+    public ApplyResult apply(SubscriptionSnapshotMessage message) {
+        try {
+            if (message == null || message.getSchemaVersion() != 1 || !isRevision(message.getRevision()))
+                return new ApplyResult(APPLY_INVALID, "消息版本或 revision 非法");
+            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, "快照时间早于当前版本");
+            }
+            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);
+
+            Path snapshots = root.resolve("snapshots");
+            Files.createDirectories(snapshots);
+            Path 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());
+            manifest.put("payload", payload);
+            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);
+            Path pointerTmp = root.resolve("current-revision.tmp");
+            Files.writeString(pointerTmp, message.getRevision() + "\n", StandardCharsets.UTF_8,
+                    StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+            atomicMove(pointerTmp, root.resolve("current-revision"));
+
+            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());
+        }
+    }
+
+    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);
+        SubscriptionSnapshotPayload payload = objectMapper.treeToValue(manifest.path("payload"), SubscriptionSnapshotPayload.class);
+        SnapshotData data = validatePayload(payload);
+        if (!isRevision(revision) || !revision.equals(sha256(objectMapper.writeValueAsBytes(payload))))
+            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-")).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 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

@@ -4,6 +4,7 @@ import lion.Domain.GalleryTask;
 import lion.Message.*;
 import lion.Message.Main.*;
 import lion.Service.DeleteService;
+import lion.Service.SubscriptionSnapshotStore;
 import lion.Service.DownloadCheckService;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
@@ -37,9 +38,19 @@ public class storageNode {
 
     ReentrantLock lock;
 
+    final SubscriptionSnapshotStore subscriptionSnapshotStore;
+
+    final ExecutorService subscriptionApplyExecutor;
+
     public static String storagePath = "/root/gallery/gallery/";
 
-    public storageNode(){
+    public storageNode(SubscriptionSnapshotStore subscriptionSnapshotStore){
+        this.subscriptionSnapshotStore = subscriptionSnapshotStore;
+        this.subscriptionApplyExecutor = Executors.newSingleThreadExecutor(r -> {
+            Thread thread = new Thread(r, "subscription-snapshot-apply");
+            thread.setDaemon(true);
+            return thread;
+        });
         queue = new HashMap<>();
         tempQueue = new HashMap<>();
         lock = new ReentrantLock();
@@ -188,6 +199,18 @@ public class storageNode {
                     ResponseMessage responseMessage = new ResponseMessage(acm.messageId, (byte)0);
                     ctx.writeAndFlush(responseMessage);
                 }
+                case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> {
+                    SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage;
+                    if (!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());
+                    });
+                }
             }
 //
 //            //修复预览
@@ -206,4 +229,8 @@ public class storageNode {
             }
         }
     }
+
+    private static String shortRevision(String revision) {
+        return revision == null ? null : revision.substring(0, Math.min(12, revision.length()));
+    }
 }

src/main/resources/config.properties

@@ -1,2 +1,9 @@
-DouNaiV2ray=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=v2
-DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta
+# 订阅快照由主站通过 Netty 长连接推送；不要在此保存上游订阅地址或 Key。
+SubscriptionSyncEnabled=false
+SubscriptionSyncSecret=
+SubscriptionDataDir=/root/gallery/storageNode/sub
+SubscriptionMaxStaleSeconds=604800
+SubscriptionMaxPayloadBytes=52428800
+SubscriptionHttpPort=8889
+SubscriptionHttpWorkers=4
+SubscriptionSocketTimeoutMs=10000

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", 1000);
+
+        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", 1000);
+        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", 1000);
+        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); }
+}