diff --git a/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java index 0c7ccad..e3a695d 100644 --- a/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java +++ b/src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java @@ -2,9 +2,11 @@ package com.lion.lionwebsite.Message; import lombok.Data; import lombok.NoArgsConstructor; +import lombok.ToString; @Data @NoArgsConstructor +@ToString(exclude = {"payloadBase64", "signature"}) public class SubscriptionSnapshotMessage extends AbstractMessage { { messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE; diff --git a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java index f195166..f7e9cd2 100644 --- a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java +++ b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java @@ -12,11 +12,13 @@ import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import io.netty.handler.logging.LoggingHandler; +import io.netty.handler.logging.ByteBufFormat; import io.netty.util.concurrent.DefaultPromise; import io.netty.util.concurrent.Promise; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import jakarta.annotation.PostConstruct; @@ -80,6 +82,8 @@ public class RemoteService { final AtomicBoolean subscriptionSyncQueued = new AtomicBoolean(); + final AtomicBoolean subscriptionSyncRunning = new AtomicBoolean(); + @Value("${subscription.standby.sync-enabled:false}") boolean subscriptionSyncEnabled; @@ -104,7 +108,8 @@ public class RemoteService { protected void initChannel(NioSocketChannel channel) { channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(100000000, 1, 4)); channel.pipeline().addLast(new MessageCodec()); - channel.pipeline().addLast(new LoggingHandler()); + // 只记录事件和字节数，避免把订阅正文、签名等消息内容写入日志。 + channel.pipeline().addLast(new LoggingHandler(io.netty.handler.logging.LogLevel.DEBUG, ByteBufFormat.SIMPLE)); channel.pipeline().addLast(new MyChannelInboundHandlerAdapter()); } }).connect(new InetSocketAddress(ip, port + i)).sync(); @@ -176,35 +181,59 @@ public class RemoteService { public void requestSubscriptionSync() { if (!subscriptionSyncEnabled) return; - if (!subscriptionSyncQueued.compareAndSet(false, true)) + subscriptionSyncQueued.set(true); + if (!subscriptionSyncRunning.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()); + subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue); + } + + @Scheduled(fixedDelayString = "${subscription.standby.retry-interval-ms:60000}") + void scheduledSubscriptionSync() { + requestSubscriptionSync(); + } + + private void drainSubscriptionSyncQueue() { + try { + while (subscriptionSyncQueued.getAndSet(false)) { + if (isDead()) + continue; + syncSubscriptionSnapshotOnce(); } - }); + } finally { + subscriptionSyncRunning.set(false); + if (subscriptionSyncQueued.get()) + requestSubscriptionSync(); + } + } + + private void syncSubscriptionSnapshotOnce() { + SubscriptionSnapshotMessage message = null; + DefaultPromise promise = null; + try { + message = subscriptionStandbySnapshotService.build(); + message.setMessageId(atomicInteger.getAndIncrement()); + 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 || response.getResult() == 3)) + log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), response.getResult()); + else + log.warn("订阅快照同步失败 revision={} result={}", shortRevision(message.getRevision()), + reply instanceof ResponseMessage response ? response.getResult() : "invalid-response"); + } else { + log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision())); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("订阅快照同步线程被中断"); + } catch (Exception e) { + log.warn("生成或发送订阅快照失败: {}", e.getMessage()); + } finally { + if (message != null && promise != null) + promiseHashMap.remove(message.messageId, promise); + } } private static String shortRevision(String revision) { diff --git a/src/main/java/com/lion/lionwebsite/Service/SubService.java b/src/main/java/com/lion/lionwebsite/Service/SubService.java index 6f4dc49..7ff848d 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubService.java @@ -23,6 +23,8 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; import java.util.Date; +import java.util.concurrent.locks.Lock; +import java.util.function.Supplier; @Service @Slf4j @@ -32,8 +34,13 @@ public class SubService { final UserMapper userMapper; final SubscriptionRefreshService refreshService; final RemoteService remoteService; + final SubscriptionStateCoordinator stateCoordinator; public String insertSubscriptionAccount(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) { + return withWriteLock(() -> insertSubscriptionAccountUnlocked(name, upstreamKey, filterHighMultiplier, enabled)); + } + + private String insertSubscriptionAccountUnlocked(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) { Response response = Response.generateResponse(); if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank()) return response.failure("名称和上游 key 不能为空").toJSONString(); @@ -57,6 +64,10 @@ public class SubService { } public String updateSubscriptionAccount(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) { + return withWriteLock(() -> updateSubscriptionAccountUnlocked(id, name, upstreamKey, filterHighMultiplier, enabled)); + } + + private String updateSubscriptionAccountUnlocked(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) { Response response = Response.generateResponse(); SubscriptionAccount account = subMapper.selectSubscriptionAccount(id); if (account == null) @@ -79,6 +90,10 @@ public class SubService { } public String deleteSubscriptionAccount(Integer id) { + return withWriteLock(() -> deleteSubscriptionAccountUnlocked(id)); + } + + private String deleteSubscriptionAccountUnlocked(Integer id) { Response response = Response.generateResponse(); SubscriptionAccount account = subMapper.selectSubscriptionAccount(id); if (account == null) @@ -91,12 +106,20 @@ public class SubService { } public String refreshSubscriptionAccount(Integer id) { + return withWriteLock(() -> refreshSubscriptionAccountUnlocked(id)); + } + + private String refreshSubscriptionAccountUnlocked(Integer id) { boolean success = refreshService.refresh(id); remoteService.requestSubscriptionSync(); return success ? Response._success("刷新成功") : Response._failure("刷新失败，请查看子账号错误状态"); } public String insertSubBind(String user, Integer accountId) { + return withWriteLock(() -> insertSubBindUnlocked(user, accountId)); + } + + private String insertSubBindUnlocked(String user, Integer accountId) { Response response = Response.generateResponse(); if (user == null || user.isBlank() || userMapper.selectUserByUsername(user) == null) return response.failure("用户不存在").toJSONString(); @@ -116,6 +139,10 @@ public class SubService { } public String resetKey(String user) { + return withWriteLock(() -> resetKeyUnlocked(user)); + } + + private String resetKeyUnlocked(String user) { Response response = Response.generateResponse(); if (subMapper.countSubBindByUser(user) == 0) return response.failure("绑定不存在").toJSONString(); @@ -129,6 +156,10 @@ public class SubService { } public String rebind(String user, Integer accountId) { + return withWriteLock(() -> rebindUnlocked(user, accountId)); + } + + private String rebindUnlocked(String user, Integer accountId) { Response response = Response.generateResponse(); SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId); if (account == null || !account.isEnabled()) @@ -193,12 +224,26 @@ public class SubService { } public String deleteSubBind(String user) { + return withWriteLock(() -> deleteSubBindUnlocked(user)); + } + + private String deleteSubBindUnlocked(String user) { subMapper.deleteSubBind(user); subMapper.deleteSubUpdateRecord(user); remoteService.requestSubscriptionSync(); return Response._success("删除成功"); } + private String withWriteLock(Supplier action) { + Lock stateLock = stateCoordinator.writeLock(); + stateLock.lock(); + try { + return action.get(); + } finally { + stateLock.unlock(); + } + } + private SubscriptionAccount firstEnabledAccount() { return subMapper.selectAllSubscriptionAccounts().stream().filter(SubscriptionAccount::isEnabled).findFirst().orElse(null); } diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java index d938bbb..eaa810d 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java @@ -19,6 +19,7 @@ import java.nio.file.*; import java.util.*; import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.concurrent.locks.Lock; @Service @Slf4j @@ -29,6 +30,8 @@ public class SubscriptionRefreshService { final SubMapper subMapper; + final SubscriptionStateCoordinator stateCoordinator; + @Value("${subscription.upstream.v2-url-template}") String v2UrlTemplate; @@ -48,6 +51,9 @@ public class SubscriptionRefreshService { } public boolean refresh(Integer accountId) { + Lock stateLock = stateCoordinator.writeLock(); + stateLock.lock(); + try { SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId); if (account == null || !account.isEnabled()) return false; @@ -66,6 +72,9 @@ public class SubscriptionRefreshService { log.error("刷新子账号订阅失败 accountId={}: {}", accountId, message); return false; } + } finally { + stateLock.unlock(); + } } public String v2Url(SubscriptionAccount account) { @@ -204,11 +213,15 @@ public class SubscriptionRefreshService { } public void invalidateCache(Integer accountId) { + Lock stateLock = stateCoordinator.writeLock(); + stateLock.lock(); try { Files.deleteIfExists(cachedPath(accountId, "v2")); Files.deleteIfExists(cachedPath(accountId, "cat")); } catch (IOException e) { log.warn("清理失效订阅缓存失败 accountId={}", accountId, e); + } finally { + stateLock.unlock(); } } } diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java index 903ce3b..ebb4c7c 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java @@ -9,6 +9,7 @@ import com.lion.lionwebsite.Util.CustomUtil; import lombok.RequiredArgsConstructor; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import jakarta.annotation.PostConstruct; import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; @@ -20,18 +21,32 @@ import java.nio.file.Path; import java.security.MessageDigest; import java.util.*; import java.util.zip.GZIPOutputStream; +import java.util.concurrent.locks.Lock; @Service @RequiredArgsConstructor public class SubscriptionStandbySnapshotService { private final SubMapper subMapper; private final SubscriptionRefreshService refreshService; + private final SubscriptionStateCoordinator stateCoordinator; private final ObjectMapper objectMapper = CustomUtil.objectMapper; @Value("${subscription.standby.sync-secret:}") String syncSecret; + @Value("${subscription.standby.sync-enabled:false}") + boolean syncEnabled; + + @PostConstruct + void validateConfiguration() { + if (syncEnabled && (syncSecret == null || syncSecret.isBlank())) + throw new IllegalStateException("启用订阅备机同步时必须配置 SUBSCRIPTION_SYNC_SECRET"); + } + public SubscriptionSnapshotMessage build() throws IOException { + Lock stateLock = stateCoordinator.readLock(); + stateLock.lock(); + try { Map accountMap = new HashMap<>(); for (SubscriptionAccount account : subMapper.selectAllSubscriptionAccounts()) { if (!account.isEnabled() || !refreshService.hasCompleteCache(account.getId())) @@ -83,6 +98,9 @@ public class SubscriptionStandbySnapshotService { message.setPayloadSha256(payloadSha256); message.setSignature(hmac(signatureInput.getBytes(StandardCharsets.UTF_8))); return message; + } finally { + stateLock.unlock(); + } } private String hmac(byte[] input) { diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStateCoordinator.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStateCoordinator.java new file mode 100644 index 0000000..eed5b06 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStateCoordinator.java @@ -0,0 +1,20 @@ +package com.lion.lionwebsite.Service; + +import org.springframework.stereotype.Component; + +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** Coordinates database bindings and the two cache files as one subscription state. */ +@Component +public final class SubscriptionStateCoordinator { + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + + public Lock readLock() { + return lock.readLock(); + } + + public Lock writeLock() { + return lock.writeLock(); + } +} diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index c778412..e9703b9 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -46,6 +46,7 @@ subscription: standby: sync-enabled: "${SUBSCRIPTION_STANDBY_SYNC_ENABLED:false}" sync-secret: "${SUBSCRIPTION_SYNC_SECRET:}" + retry-interval-ms: 60000 bot: token: "5222939329:AAHa6l9ZuVVdNSDLPI_H-c8O_VgeOEw5plA" diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java new file mode 100644 index 0000000..de252f3 --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java @@ -0,0 +1,63 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Dao.normal.SubMapper; +import com.lion.lionwebsite.Domain.SubBind; +import com.lion.lionwebsite.Domain.SubscriptionAccount; +import com.lion.lionwebsite.Message.SubscriptionSnapshotMessage; +import com.lion.lionwebsite.Message.SubscriptionSnapshotPayload; +import com.lion.lionwebsite.Util.CustomUtil; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Base64; +import java.util.zip.GZIPInputStream; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +class SubscriptionStandbySnapshotServiceTest { + @Test + void buildsDeterministicSnapshotWithoutPlainPublicKeys(@TempDir Path directory) throws Exception { + SubMapper mapper = mock(SubMapper.class); + SubscriptionRefreshService refreshService = mock(SubscriptionRefreshService.class); + SubscriptionStateCoordinator coordinator = new SubscriptionStateCoordinator(); + SubscriptionAccount account = new SubscriptionAccount(1, "account", "upstream-secret", true, true, + null, null, null, null, 2, null, null); + when(mapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(java.util.List.of(account))); + when(mapper.selectAllSubBind()).thenReturn(new ArrayList<>(java.util.List.of( + new SubBind("public-key-a", "user-a", 1, "account", true, true), + new SubBind("public-key-b", "user-b", 1, "account", true, true)))); + Path v2 = directory.resolve("v2ray.txt"); + Path clash = directory.resolve("clash.yaml"); + Files.writeString(v2, "v2-content", StandardCharsets.UTF_8); + Files.writeString(clash, "clash-content", StandardCharsets.UTF_8); + when(refreshService.hasCompleteCache(1)).thenReturn(true); + when(refreshService.cachedPath(1, "v2")).thenReturn(v2); + when(refreshService.cachedPath(1, "cat")).thenReturn(clash); + + SubscriptionStandbySnapshotService service = new SubscriptionStandbySnapshotService(mapper, refreshService, coordinator); + service.syncSecret = "test-sync-secret"; + SubscriptionSnapshotMessage first = service.build(); + SubscriptionSnapshotMessage second = service.build(); + + assertEquals(first.getRevision(), second.getRevision()); + byte[] compressed = Base64.getDecoder().decode(first.getPayloadBase64()); + byte[] json; + try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) { + json = gzip.readAllBytes(); + } + String rawPayload = new String(json, StandardCharsets.UTF_8); + assertFalse(rawPayload.contains("public-key-a")); + assertFalse(rawPayload.contains("public-key-b")); + assertFalse(rawPayload.contains("upstream-secret")); + SubscriptionSnapshotPayload payload = CustomUtil.objectMapper.readValue(json, SubscriptionSnapshotPayload.class); + assertEquals(1, payload.getAccounts().size()); + assertEquals(2, payload.getBindings().size()); + assertTrue(payload.getAccounts().getFirst().isFilterHighMultiplier()); + } +}