增加订阅快照备机同步链路

2026-08-30 10:24:57 +08:00
parent fc8548f90f
commit 14dd12d9ad
12 changed files with 259 additions and 2 deletions
@@ -21,7 +21,9 @@ import org.springframework.context.annotation.Configuration;
        SendResponse.class, Message.class, com.pengrad.telegrambot.model.User.class,
        Chat.class, MessageEntity.class,
        AbstractMethodError.class, DeleteGalleryMessage.class, DownloadPostMessage.class, DownloadStatusMessage.class,
-        IdentityMessage.class, MaintainMessage.class, ResponseMessage.class, AvailableCheckMessage.class, LinkPreviewOptions.class})
+        IdentityMessage.class, MaintainMessage.class, ResponseMessage.class, AvailableCheckMessage.class,
+        SubscriptionSnapshotMessage.class, SubscriptionSnapshotPayload.class, SubscriptionAccountSnapshot.class,
+        SubscriptionBindingSnapshot.class, LinkPreviewOptions.class})
 public class CustomBean {

    @Value("${bot.token:5222939329:AAHa6l9ZuVVdNSDLPI_H-c8O_VgeOEw5plA}")
@@ -19,6 +19,10 @@ public class AbstractMessage {

    public static final byte AVAILABLE_CHECK_MESSAGE = 8;

+    public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
+
+    public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
+
    public byte messageType;

    public int messageId;
@@ -44,6 +44,8 @@ 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);
+            case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> objectMapper.readValue(metadata, SubscriptionSnapshotMessage.class);
            default -> null;
        };

@@ -0,0 +1,16 @@
+package com.lion.lionwebsite.Message;
+
+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;
+}
@@ -0,0 +1,11 @@
+package com.lion.lionwebsite.Message;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionBindingSnapshot {
+    private String publicKeySha256;
+    private Integer accountId;
+}
@@ -0,0 +1,19 @@
+package com.lion.lionwebsite.Message;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@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;
+}
@@ -0,0 +1,15 @@
+package com.lion.lionwebsite.Message;
+
+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<>();
+}
@@ -118,6 +118,7 @@ public class LocalService{
    public boolean updateSub(boolean isManual) throws IOException {
        // 手动和定时入口均刷新全部启用子账号；全部成功后更新原有“上次更新时间”。
        boolean success = subscriptionRefreshService.refreshAll();
+        remoteService.requestSubscriptionSync();
        if (success)
            configurationMapper.updateConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME,
                    CustomUtil.dateTimeFormatter().format(LocalDateTime.now()));
@@ -20,6 +20,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;

 import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;

 import java.io.IOException;
 import java.io.OutputStream;
@@ -34,6 +35,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicBoolean;

@Service
@Data
@@ -68,6 +70,19 @@ public class RemoteService {

    final WebSocketService webSocketService;

+    final SubscriptionStandbySnapshotService subscriptionStandbySnapshotService;
+
+    final ExecutorService subscriptionSyncExecutor = Executors.newSingleThreadExecutor(r -> {
+        Thread thread = new Thread(r, "subscription-standby-sync");
+        thread.setDaemon(true);
+        return thread;
+    });
+
+    final AtomicBoolean subscriptionSyncQueued = new AtomicBoolean();
+
+    @Value("${subscription.standby.sync-enabled:false}")
+    boolean subscriptionSyncEnabled;
+
    @PostConstruct
    void init() {
        if(!initChannel()){     //如果远程服务器连接失败，则开启本地监听
@@ -113,6 +128,7 @@ public class RemoteService {

            //子节点上线时，发送未完成的任务
            resetUndone();
+            requestSubscriptionSync();
            return true;
        }catch (Exception e){
            log.error("connect node failed, wait for node back online", e);
@@ -156,6 +172,50 @@ public class RemoteService {
        }
    }

+    /** 请求将当前全部订阅状态异步同步到存储节点，短时间内的多次请求会合并。 */
+    public void requestSubscriptionSync() {
+        if (!subscriptionSyncEnabled)
+            return;
+        if (!subscriptionSyncQueued.compareAndSet(false, true))
+            return;
+        subscriptionSyncExecutor.execute(() -> {
+            try {
+                do {
+                    subscriptionSyncQueued.set(false);
+                    if (isDead())
+                        continue;
+                    SubscriptionSnapshotMessage message = subscriptionStandbySnapshotService.build();
+                    message.setMessageId(atomicInteger.getAndIncrement());
+                    DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
+                    promiseHashMap.put(message.messageId, promise);
+                    channel.writeAndFlush(message);
+                    if (promise.await(30, TimeUnit.SECONDS)) {
+                        AbstractMessage reply = promise.getNow();
+                        if (reply instanceof ResponseMessage response && response.getResult() == 0)
+                            log.info("订阅快照同步成功 revision={}", shortRevision(message.getRevision()));
+                        else
+                            log.warn("订阅快照同步失败 revision={} result={}", shortRevision(message.getRevision()),
+                                    reply instanceof ResponseMessage response ? response.getResult() : "timeout");
+                    } else {
+                        log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision()));
+                    }
+                    promiseHashMap.remove(message.messageId, promise);
+                } while (subscriptionSyncQueued.getAndSet(false));
+            } catch (Exception e) {
+                log.warn("生成或发送订阅快照失败: {}", e.getMessage());
+            }
+        });
+    }
+
+    private static String shortRevision(String revision) {
+        return revision == null ? null : revision.substring(0, Math.min(12, revision.length()));
+    }
+
+    @PreDestroy
+    void shutdownSubscriptionSync() {
+        subscriptionSyncExecutor.shutdownNow();
+    }
+
    public boolean isDead(){
        return channelFuture == null || channelFuture.channel() == null || !channelFuture.channel().isActive();
    }
@@ -31,6 +31,7 @@ public class SubService {
    final SubMapper subMapper;
    final UserMapper userMapper;
    final SubscriptionRefreshService refreshService;
+    final RemoteService remoteService;

    public String insertSubscriptionAccount(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
        Response response = Response.generateResponse();
@@ -42,6 +43,7 @@ public class SubService {
        subMapper.insertSubscriptionAccount(account);
        if (enabled)
            refreshService.refresh(account.getId());
+        remoteService.requestSubscriptionSync();
        return response.success(accountJson(account)).toJSONString();
    }

@@ -72,6 +74,7 @@ public class SubService {
        subMapper.updateSubscriptionAccount(account);
        if (enabled && !refreshService.refresh(id))
            refreshService.invalidateCache(id);
+        remoteService.requestSubscriptionSync();
        return response.success(accountJson(account)).toJSONString();
    }

@@ -83,11 +86,14 @@ public class SubService {
        if (account.getBoundUserCount() != null && account.getBoundUserCount() > 0)
            return response.failure("子账号仍绑定用户，请先改绑").toJSONString();
        subMapper.deleteSubscriptionAccount(id);
+        remoteService.requestSubscriptionSync();
        return response.success("删除成功").toJSONString();
    }

    public String refreshSubscriptionAccount(Integer id) {
-        return refreshService.refresh(id) ? Response._success("刷新成功") : Response._failure("刷新失败，请查看子账号错误状态");
+        boolean success = refreshService.refresh(id);
+        remoteService.requestSubscriptionSync();
+        return success ? Response._success("刷新成功") : Response._failure("刷新失败，请查看子账号错误状态");
    }

    public String insertSubBind(String user, Integer accountId) {
@@ -105,6 +111,7 @@ public class SubService {
        while (subMapper.selectSubBindExist(key))
            key = RandomUtil.randomString(8);
        subMapper.insertSubBind(new SubBind(key, user, account.getId(), account.getName(), account.isEnabled(), account.isFilterHighMultiplier()));
+        remoteService.requestSubscriptionSync();
        return response.success("添加成功").toJSONString();
    }

@@ -117,6 +124,7 @@ public class SubService {
            key = RandomUtil.randomString(8);
        subMapper.updateSubBindKey(user, key);
        subMapper.deleteSubUpdateRecord(user);
+        remoteService.requestSubscriptionSync();
        return response.success().toJSONString();
    }

@@ -129,6 +137,7 @@ public class SubService {
            return response.failure("子账号尚无有效缓存，请先刷新").toJSONString();
        if (subMapper.updateSubBindAccount(user, accountId) == 0)
            return response.failure("绑定不存在").toJSONString();
+        remoteService.requestSubscriptionSync();
        return response.success("改绑成功").toJSONString();
    }

@@ -186,6 +195,7 @@ public class SubService {
    public String deleteSubBind(String user) {
        subMapper.deleteSubBind(user);
        subMapper.deleteSubUpdateRecord(user);
+        remoteService.requestSubscriptionSync();
        return Response._success("删除成功");
    }

@@ -0,0 +1,114 @@
+package com.lion.lionwebsite.Service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.lion.lionwebsite.Dao.normal.SubMapper;
+import com.lion.lionwebsite.Domain.SubBind;
+import com.lion.lionwebsite.Domain.SubscriptionAccount;
+import com.lion.lionwebsite.Message.*;
+import com.lion.lionwebsite.Util.CustomUtil;
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.zip.GZIPOutputStream;
+
+@Service
+@RequiredArgsConstructor
+public class SubscriptionStandbySnapshotService {
+    private final SubMapper subMapper;
+    private final SubscriptionRefreshService refreshService;
+    private final ObjectMapper objectMapper = CustomUtil.objectMapper;
+
+    @Value("${subscription.standby.sync-secret:}")
+    String syncSecret;
+
+    public SubscriptionSnapshotMessage build() throws IOException {
+        Map<Integer, SubscriptionAccountSnapshot> accountMap = new HashMap<>();
+        for (SubscriptionAccount account : subMapper.selectAllSubscriptionAccounts()) {
+            if (!account.isEnabled() || !refreshService.hasCompleteCache(account.getId()))
+                continue;
+            Path v2Path = refreshService.cachedPath(account.getId(), "v2");
+            Path clashPath = refreshService.cachedPath(account.getId(), "cat");
+            byte[] v2 = Files.readAllBytes(v2Path);
+            byte[] clash = Files.readAllBytes(clashPath);
+            SubscriptionAccountSnapshot snapshot = new SubscriptionAccountSnapshot();
+            snapshot.setAccountId(account.getId());
+            snapshot.setEnabled(true);
+            snapshot.setFilterHighMultiplier(account.isFilterHighMultiplier());
+            snapshot.setV2ContentBase64(Base64.getEncoder().encodeToString(v2));
+            snapshot.setV2Sha256(sha256(v2));
+            snapshot.setClashContentBase64(Base64.getEncoder().encodeToString(clash));
+            snapshot.setClashSha256(sha256(clash));
+            accountMap.put(account.getId(), snapshot);
+        }
+
+        List<SubscriptionAccountSnapshot> accounts = new ArrayList<>(accountMap.values());
+        accounts.sort(Comparator.comparing(SubscriptionAccountSnapshot::getAccountId));
+        List<SubscriptionBindingSnapshot> bindings = new ArrayList<>();
+        for (SubBind bind : subMapper.selectAllSubBind()) {
+            if (bind.getSubscriptionAccountId() == null || !accountMap.containsKey(bind.getSubscriptionAccountId()))
+                continue;
+            SubscriptionBindingSnapshot snapshot = new SubscriptionBindingSnapshot();
+            snapshot.setPublicKeySha256(sha256(bind.getKey().getBytes(StandardCharsets.UTF_8)));
+            snapshot.setAccountId(bind.getSubscriptionAccountId());
+            bindings.add(snapshot);
+        }
+        bindings.sort(Comparator.comparing(SubscriptionBindingSnapshot::getPublicKeySha256));
+
+        SubscriptionSnapshotPayload payload = new SubscriptionSnapshotPayload();
+        payload.setSchemaVersion(1);
+        payload.setAccounts(accounts);
+        payload.setBindings(bindings);
+        byte[] payloadJson = objectMapper.writeValueAsBytes(payload);
+        byte[] compressed = gzip(payloadJson);
+        String revision = sha256(payloadJson);
+        String payloadSha256 = sha256(compressed);
+        long generatedAt = System.currentTimeMillis();
+        String signatureInput = "1\n" + revision + "\n" + generatedAt + "\n" + payloadSha256;
+
+        SubscriptionSnapshotMessage message = new SubscriptionSnapshotMessage();
+        message.setSchemaVersion(1);
+        message.setRevision(revision);
+        message.setGeneratedAt(generatedAt);
+        message.setPayloadBase64(Base64.getEncoder().encodeToString(compressed));
+        message.setPayloadSha256(payloadSha256);
+        message.setSignature(hmac(signatureInput.getBytes(StandardCharsets.UTF_8)));
+        return message;
+    }
+
+    private String hmac(byte[] input) {
+        if (syncSecret == null || syncSecret.isBlank())
+            throw new IllegalStateException("订阅备机同步密钥未配置");
+        try {
+            Mac mac = Mac.getInstance("HmacSHA256");
+            mac.init(new SecretKeySpec(syncSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+            return hex(mac.doFinal(input));
+        } catch (Exception e) {
+            throw new IllegalStateException("生成订阅快照签名失败", e);
+        }
+    }
+
+    private static byte[] gzip(byte[] input) throws IOException {
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        try (GZIPOutputStream gzip = new GZIPOutputStream(output)) {
+            gzip.write(input);
+        }
+        return output.toByteArray();
+    }
+
+    private static String sha256(byte[] input) {
+        try { return hex(MessageDigest.getInstance("SHA-256").digest(input)); }
+        catch (Exception e) { throw new IllegalStateException(e); }
+    }
+
+    private static String hex(byte[] input) { return HexFormat.of().formatHex(input); }
+}
@@ -43,6 +43,9 @@ subscription:
    clash-url-template: "https://aaaa.gay/link/{key}?client=clashmeta"
    high-multiplier-threshold: 2.0
    refresh-interval-ms: 86400000
+  standby:
+    sync-enabled: "${SUBSCRIPTION_STANDBY_SYNC_ENABLED:false}"
+    sync-secret: "${SUBSCRIPTION_SYNC_SECRET:}"

 bot:
  token: "5222939329:AAHa6l9ZuVVdNSDLPI_H-c8O_VgeOEw5plA"