diff --git a/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java b/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java index 8124ab8..0f48f4c 100644 --- a/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java +++ b/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java @@ -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}") diff --git a/src/main/java/com/lion/lionwebsite/Message/AbstractMessage.java b/src/main/java/com/lion/lionwebsite/Message/AbstractMessage.java index db1ed86..6f0d449 100644 --- a/src/main/java/com/lion/lionwebsite/Message/AbstractMessage.java +++ b/src/main/java/com/lion/lionwebsite/Message/AbstractMessage.java @@ -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; diff --git a/src/main/java/com/lion/lionwebsite/Message/MessageCodec.java b/src/main/java/com/lion/lionwebsite/Message/MessageCodec.java index 1126b82..08025f2 100644 --- a/src/main/java/com/lion/lionwebsite/Message/MessageCodec.java +++ b/src/main/java/com/lion/lionwebsite/Message/MessageCodec.java @@ -44,6 +44,8 @@ public class MessageCodec extends ByteToMessageCodec { 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; }; diff --git a/src/main/java/com/lion/lionwebsite/Message/SubscriptionAccountSnapshot.java b/src/main/java/com/lion/lionwebsite/Message/SubscriptionAccountSnapshot.java new file mode 100644 index 0000000..f0a0551 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Message/SubscriptionAccountSnapshot.java @@ -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; +} diff --git a/src/main/java/com/lion/lionwebsite/Message/SubscriptionBindingSnapshot.java b/src/main/java/com/lion/lionwebsite/Message/SubscriptionBindingSnapshot.java new file mode 100644 index 0000000..74bed94 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Message/SubscriptionBindingSnapshot.java @@ -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; +} diff --git a/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java new file mode 100644 index 0000000..0c7ccad --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java @@ -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; +} diff --git a/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotPayload.java b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotPayload.java new file mode 100644 index 0000000..4259c56 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotPayload.java @@ -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 accounts = new ArrayList<>(); + private List bindings = new ArrayList<>(); +} diff --git a/src/main/java/com/lion/lionwebsite/Service/LocalService.java b/src/main/java/com/lion/lionwebsite/Service/LocalService.java index 6f10f88..4d12a7e 100644 --- a/src/main/java/com/lion/lionwebsite/Service/LocalService.java +++ b/src/main/java/com/lion/lionwebsite/Service/LocalService.java @@ -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())); diff --git a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java index a17f5e1..f195166 100644 --- a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java +++ b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java @@ -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 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(); } diff --git a/src/main/java/com/lion/lionwebsite/Service/SubService.java b/src/main/java/com/lion/lionwebsite/Service/SubService.java index c81810c..6f4dc49 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubService.java @@ -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("删除成功"); } diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java new file mode 100644 index 0000000..903ce3b --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java @@ -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 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 accounts = new ArrayList<>(accountMap.values()); + accounts.sort(Comparator.comparing(SubscriptionAccountSnapshot::getAccountId)); + List 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); } +} diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 4e83a3d..c778412 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -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"