Record standby subscription events with signed durable receipts

src/main/java/com/lion/lionwebsite/Controller/SubscriptionEventController.java

@@ -0,0 +1,33 @@
+package com.lion.lionwebsite.Controller;
+
+import com.lion.lionwebsite.Service.SubscriptionEventService;
+import jakarta.servlet.http.HttpServletRequest;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.IOException;
+import java.util.Map;
+
+@RestController
+public class SubscriptionEventController {
+    private final SubscriptionEventService events;
+
+    public SubscriptionEventController(SubscriptionEventService events) { this.events = events; }
+
+    @PostMapping(value = "/internal/subscription-events", produces = "application/json")
+    public ResponseEntity<Map<String, Object>> ingest(HttpServletRequest request) throws IOException {
+        // Bound actual bytes even for chunked requests and dishonest Content-Length headers.
+        byte[] body = request.getInputStream().readNBytes(SubscriptionEventService.MAX_BODY + 1);
+        if (body.length > SubscriptionEventService.MAX_BODY)
+            return ResponseEntity.badRequest().body(Map.of("error", "invalid subscription events"));
+        if (!events.verify(body, request.getHeader("X-Subscription-Timestamp"), request.getHeader("X-Subscription-Signature")))
+            return ResponseEntity.status(403).body(Map.of("error", "authentication failed"));
+        try {
+            return ResponseEntity.ok(events.ingest(events.parseBatch(body)));
+        } catch (IllegalArgumentException invalid) {
+            return ResponseEntity.badRequest().body(Map.of("error", "invalid subscription events"));
+        }
+        // All infrastructure failures propagate as 5xx; never acknowledge an uncommitted batch.
+    }
+}

src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java

@@ -80,7 +80,7 @@ public interface SubMapper {
     @Select("select count(user) from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user})")
     Integer selectUpdateRecordCount(String user);
 
-    @Select("select min(id) from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user})")
+    @Select("select id from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user}) order by time asc,id asc limit 1")
     Integer selectMinUpdateRecordId(String user);
 
     /** 后台补齐归属地：只更新该用户最近一条记录，避免覆盖更早的历史记录。 */

src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java

@@ -0,0 +1,31 @@
+package com.lion.lionwebsite.Dao.normal;
+
+import com.lion.lionwebsite.Domain.SubscriptionEvent;
+import com.lion.lionwebsite.Domain.SubUpdateRecord;
+import com.lion.lionwebsite.Domain.SubscriptionEventBinding;
+import org.apache.ibatis.annotations.*;
+
+@Mapper
+public interface SubscriptionEventMapper {
+    @Select("select fingerprint from subscription_event_receipt where event_id=#{eventId}")
+    String fingerprint(String eventId);
+
+    @Insert("insert into subscription_event_receipt(event_id,key_hash,client,occurred_at,source,ingested_at,fingerprint,outcome) values(#{event.eventId},#{event.keyHash},#{event.client},#{event.occurredAt},#{source},#{ingestedAt},#{fingerprint},#{outcome})")
+    void insertReceipt(@Param("event") SubscriptionEvent event, @Param("source") String source,
+                       @Param("ingestedAt") long ingestedAt, @Param("fingerprint") String fingerprint,
+                       @Param("outcome") String outcome);
+
+    @Select("select key_hash as keyHash,user_id as userId,user_name as userName from subscription_event_binding where key_hash=#{hash}")
+    SubscriptionEventBinding binding(String hash);
+
+    @Insert("insert into subscription_event_binding(key_hash,user_id,user_name,first_seen_at,last_seen_at) values(#{hash},#{userId},#{userName},#{now},#{now}) on conflict(key_hash) do update set user_name=excluded.user_name,last_seen_at=excluded.last_seen_at where subscription_event_binding.user_id IS excluded.user_id")
+    void rememberBinding(@Param("hash") String hash, @Param("userId") Integer userId,
+                         @Param("userName") String userName, @Param("now") long now);
+
+    @Insert("insert into sub_update_record(user,ip,UA,time,location,user_id) values(#{user},#{ip},#{UA},#{time},#{location},#{userId})")
+    @Options(useGeneratedKeys = true, keyProperty = "id")
+    void insertRecord(SubUpdateRecord record);
+
+    @Delete("delete from sub_update_record where (user_id=#{userId} or (user_id is null and user=#{userName})) and id not in (select id from sub_update_record where user_id=#{userId} or (user_id is null and user=#{userName}) order by time desc,id desc limit 10)")
+    void pruneRecords(@Param("userId") Integer userId, @Param("userName") String userName);
+}

src/main/java/com/lion/lionwebsite/Domain/SubscriptionEvent.java

@@ -0,0 +1,16 @@
+package com.lion.lionwebsite.Domain;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+/** Validated, authenticated event. Never contains the public subscription key. */
+@Data
+@AllArgsConstructor
+public class SubscriptionEvent {
+    private String eventId;
+    private String keyHash;
+    private String client;
+    private String clientIp;
+    private String userAgent;
+    private long occurredAt;
+}

src/main/java/com/lion/lionwebsite/Domain/SubscriptionEventBinding.java

@@ -0,0 +1,10 @@
+package com.lion.lionwebsite.Domain;
+
+import lombok.Data;
+
+@Data
+public class SubscriptionEventBinding {
+    private String keyHash;
+    private Integer userId;
+    private String userName;
+}

src/main/java/com/lion/lionwebsite/Service/SubService.java

@@ -5,6 +5,7 @@ import com.lion.lionwebsite.Dao.normal.SubMapper;
 import com.lion.lionwebsite.Dao.normal.UserMapper;
 import com.lion.lionwebsite.Domain.SubBind;
 import com.lion.lionwebsite.Domain.SubUpdateRecord;
+import com.lion.lionwebsite.Domain.SubscriptionEvent;
 import com.lion.lionwebsite.Domain.SubscriptionAccount;
 import com.lion.lionwebsite.Util.CustomUtil;
 import com.lion.lionwebsite.Util.FileDownload;
@@ -47,6 +48,9 @@ import java.util.function.Supplier;
         @Qualifier("mainTransactionTemplate")
         TransactionTemplate transactions;
 
+        @Autowired(required = false)
+        SubscriptionEventService subscriptionEvents;
+
         /**
          * 归属地查询是同步外呼（最长十几秒），且失败只影响记录里的一列展示。
          * 若放在订阅分发路径上，用户取一次订阅就得先等它完成，因此改为后台补齐。
@@ -188,6 +192,7 @@ import java.util.function.Supplier;
         String key = RandomUtil.randomString(8);
         while (subMapper.selectSubBindExist(key))
             key = RandomUtil.randomString(8);
+        rememberEventBindings();
         subMapper.updateSubBindKey(user, key);
         subMapper.deleteSubUpdateRecord(user);
         remoteService.requestSubscriptionSync();
@@ -239,8 +244,9 @@ import java.util.function.Supplier;
         }
         String ip = resolveClientIp(request);
         String ua = request.getHeader("User-Agent");
-        if (ua == null) return;
-        recordUpdate(subBind, ip, ua);
+        SubscriptionEvent event = subscriptionEvents == null ? null : subscriptionEvents.requestEvent(request, key, client);
+        if (ua == null && event == null) return;
+        if (event == null) recordUpdate(subBind, ip, ua);
         if (!"v2".equals(client) && !"cat".equals(client)) {
             sendStatus(response, HttpServletResponse.SC_BAD_REQUEST, "client error");
             return;
@@ -250,6 +256,8 @@ import java.util.function.Supplier;
             sendStatus(response, HttpServletResponse.SC_SERVICE_UNAVAILABLE, "subscription unavailable");
             return;
         }
+        if (event != null && !"HEAD".equalsIgnoreCase(request.getMethod()))
+            subscriptionEvents.recordPrimary(event, subBind);
         FileDownload.export(request, response, path.toString());
     }
 
@@ -329,12 +337,17 @@ import java.util.function.Supplier;
     }
 
     private String deleteSubBindUnlocked(String user) {
+        rememberEventBindings();
         subMapper.deleteSubBind(user);
         subMapper.deleteSubUpdateRecord(user);
         remoteService.requestSubscriptionSync();
         return Response._success("删除成功");
     }
 
+    private void rememberEventBindings() {
+        if (subscriptionEvents != null) subscriptionEvents.rememberBindings(subMapper.selectAllSubBind());
+    }
+
     private String withWriteLock(Supplier<String> action) {
         Lock stateLock = stateCoordinator.writeLock();
         stateLock.lock();

src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java

@@ -0,0 +1,222 @@
+package com.lion.lionwebsite.Service;
+
+import com.lion.lionwebsite.Dao.normal.SubMapper;
+import com.lion.lionwebsite.Dao.normal.SubscriptionEventMapper;
+import com.lion.lionwebsite.Dao.normal.UserMapper;
+import com.lion.lionwebsite.Domain.*;
+import com.lion.lionwebsite.Util.CustomUtil;
+import jakarta.annotation.PreDestroy;
+import jakarta.servlet.http.HttpServletRequest;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
+import tools.jackson.databind.JsonNode;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.time.Clock;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.locks.Lock;
+
+@Service
+public class SubscriptionEventService {
+    public static final int MAX_BODY = 256 * 1024;
+    private final SubscriptionEventMapper events;
+    private final SubMapper subscriptions;
+    private final UserMapper users;
+    private final TransactionTemplate transactions;
+    private final SubscriptionStateCoordinator coordinator;
+    @Value("${subscription.standby.sync-secret:}") String syncSecret;
+    Clock clock = Clock.systemUTC();
+    private final ExecutorService locations = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+            new ArrayBlockingQueue<>(128), runnable -> {
+        Thread thread = new Thread(runnable, "subscription-event-location");
+        thread.setDaemon(true);
+        return thread;
+    }, new ThreadPoolExecutor.AbortPolicy());
+
+    public SubscriptionEventService(SubscriptionEventMapper events, SubMapper subscriptions, UserMapper users,
+                                    @Qualifier("mainTransactionTemplate") TransactionTemplate transactions,
+                                    SubscriptionStateCoordinator coordinator) {
+        this.events = events;
+        this.subscriptions = subscriptions;
+        this.users = users;
+        this.transactions = transactions;
+        this.coordinator = coordinator;
+    }
+
+    /** Auth failure deliberately discloses neither signature nor payload. */
+    public boolean verify(byte[] body, String timestamp, String signature) {
+        if (body == null || body.length > MAX_BODY || syncSecret == null || syncSecret.isBlank()
+                || timestamp == null || !timestamp.matches("[0-9]{1,12}")
+                || signature == null || !signature.matches("[0-9a-fA-F]{64}")) return false;
+        try {
+            long seconds = Long.parseLong(timestamp);
+            if (Math.abs(clock.instant().getEpochSecond() - seconds) > 300) return false;
+            Mac mac = Mac.getInstance("HmacSHA256");
+            mac.init(new SecretKeySpec(syncSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+            byte[] expected = mac.doFinal(("v1\n" + timestamp + "\n" + sha256(body)).getBytes(StandardCharsets.UTF_8));
+            return MessageDigest.isEqual(expected, HexFormat.of().parseHex(signature));
+        } catch (Exception invalid) { return false; }
+    }
+
+    /** Invalid or absent signed headers fall back to the ordinary logging path. */
+    public SubscriptionEvent requestEvent(HttpServletRequest request, String key, String client) {
+        String encoded = request.getHeader("X-Subscription-Event");
+        if (encoded == null || encoded.length() > 8192 || !encoded.matches("[A-Za-z0-9_-]+")) return null;
+        try {
+            byte[] body = Base64.getUrlDecoder().decode(encoded);
+            if (!verify(body, request.getHeader("X-Subscription-Timestamp"), request.getHeader("X-Subscription-Signature"))) return null;
+            SubscriptionEvent event = parse(CustomUtil.objectMapper.readTree(body));
+            String ua = Objects.requireNonNullElse(request.getHeader("User-Agent"), "");
+            if (!event.getKeyHash().equals(sha256(key.getBytes(StandardCharsets.UTF_8)))
+                    || !event.getClient().equals(client) || !event.getUserAgent().equals(ua)) return null;
+            return event;
+        } catch (Exception invalid) { return null; }
+    }
+
+    public List<SubscriptionEvent> parseBatch(byte[] body) {
+        try {
+            if (body == null || body.length > MAX_BODY) throw new IllegalArgumentException();
+            JsonNode root = CustomUtil.objectMapper.readTree(body);
+            JsonNode list = root.get("events");
+            if (!root.isObject() || list == null || !list.isArray() || list.size() > 100) throw new IllegalArgumentException();
+            List<SubscriptionEvent> result = new ArrayList<>();
+            for (JsonNode item : list) result.add(parse(item));
+            return result;
+        } catch (Exception invalid) { throw new IllegalArgumentException("invalid subscription events"); }
+    }
+
+    SubscriptionEvent parse(JsonNode node) {
+        if (node == null || !node.isObject()) throw new IllegalArgumentException();
+        String id = text(node, "eventId", 36);
+        if (!UUID.fromString(id).toString().equals(id)) throw new IllegalArgumentException();
+        String hash = text(node, "keyHash", 64);
+        if (!hash.matches("[0-9a-f]{64}")) throw new IllegalArgumentException();
+        String client = text(node, "client", 3);
+        if (!client.equals("v2") && !client.equals("cat")) throw new IllegalArgumentException();
+        String ip = text(node, "clientIp", 64);
+        // Literal IPv4/IPv6 only: no host names, URL fragments or query injection into geolocation.
+        if (ip.isEmpty() || !ip.matches("[0-9A-Fa-f:.]+")) throw new IllegalArgumentException();
+        String ua = text(node, "userAgent", 2048);
+        if (ua.chars().anyMatch(c -> c < 32 || c == 127)) throw new IllegalArgumentException();
+        JsonNode time = node.get("occurredAt");
+        if (time == null || !time.isIntegralNumber() || !time.canConvertToLong()) throw new IllegalArgumentException();
+        long occurred = time.longValue();
+        if (occurred < 0 || occurred > clock.millis() + 300_000) throw new IllegalArgumentException();
+        return new SubscriptionEvent(id, hash, client, ip, ua, occurred);
+    }
+
+    private static String text(JsonNode node, String field, int max) {
+        JsonNode value = node.get(field);
+        if (value == null || !value.isString() || value.stringValue().length() > max) throw new IllegalArgumentException();
+        return value.stringValue();
+    }
+
+    public Map<String, Object> ingest(List<SubscriptionEvent> batch) {
+        Lock lock = coordinator.readLock();
+        lock.lock();
+        try {
+            // Return acknowledgements only after execute has returned and the entire transaction committed.
+            List<String> acknowledged = transactions.execute(status -> {
+                List<String> ids = new ArrayList<>();
+                for (SubscriptionEvent event : batch) if (persist(event, "standby", null)) ids.add(event.getEventId());
+                return ids;
+            });
+            return Map.of("acknowledged", Objects.requireNonNull(acknowledged));
+        } finally { lock.unlock(); }
+    }
+
+    public void recordPrimary(SubscriptionEvent event, SubBind binding) {
+        transactions.executeWithoutResult(status -> {
+            if (!persist(event, "primary", binding)) throw new IllegalStateException("subscription user unavailable");
+        });
+    }
+
+    private boolean persist(SubscriptionEvent event, String source, SubBind current) {
+        String fingerprint = sha256((event.getEventId() + "\n" + event.getKeyHash() + "\n" + event.getClient()
+                + "\n" + event.getOccurredAt() + "\n" + event.getClientIp() + "\n" + event.getUserAgent()).getBytes(StandardCharsets.UTF_8));
+        String old = events.fingerprint(event.getEventId());
+        if (old != null) {
+            if (!old.equals(fingerprint)) throw new IllegalArgumentException("event ID conflict");
+            return true;
+        }
+        Integer userId = current == null ? null : current.getUserId();
+        String userName = current == null ? null : current.getUser();
+        if (current == null) {
+            boolean found = false;
+            for (SubBind binding : subscriptions.selectAllSubBind()) {
+                if (sha256(binding.getKey().getBytes(StandardCharsets.UTF_8)).equals(event.getKeyHash())) {
+                    userId = binding.getUserId();
+                    userName = binding.getUser();
+                    found = true;
+                    break;
+                }
+            }
+            if (!found) {
+                SubscriptionEventBinding binding = events.binding(event.getKeyHash());
+                if (binding == null) return false;
+                userId = binding.getUserId();
+                userName = binding.getUserName();
+            }
+        }
+        if (userId != null) {
+            User user = users.selectUserById(userId);
+            if (user == null) {
+                events.insertReceipt(event, source, clock.millis(), fingerprint, "user_deleted");
+                return true;
+            }
+            userName = user.getUsername();
+        }
+        // NULL user_id is a valid legacy subscription label, not a missing/deleted User.
+        // Keep it NULL even if someone later creates a User with the same label.
+        if (userName == null || userName.isBlank()) return false;
+        events.rememberBinding(event.getKeyHash(), userId, userName, clock.millis());
+        SubUpdateRecord record = new SubUpdateRecord(0, userName, event.getClientIp(), event.getUserAgent(),
+                new Date(event.getOccurredAt()), "unknown", userId);
+        events.insertRecord(record);
+        events.pruneRecords(userId, userName);
+        events.insertReceipt(event, source, clock.millis(), fingerprint, "recorded");
+        // Never perform network calls before commit. A dropped/rolled-back record has no enrichment task.
+        org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization(
+                new org.springframework.transaction.support.TransactionSynchronization() {
+                    @Override public void afterCommit() { enrich(record.getId(), event.getClientIp()); }
+                });
+        return true;
+    }
+
+    public void rememberBindings(Collection<SubBind> bindings) {
+        for (SubBind binding : bindings) {
+            Integer userId = binding.getUserId();
+            String name = binding.getUser();
+            if (userId != null) {
+                User user = users.selectUserById(userId);
+                if (user != null) name = user.getUsername();
+            }
+            if (name != null && !name.isBlank()) events.rememberBinding(
+                    sha256(binding.getKey().getBytes(StandardCharsets.UTF_8)), userId, name, clock.millis());
+        }
+    }
+
+    private void enrich(int recordId, String ip) {
+        if (recordId <= 0) return;
+        try { locations.execute(() -> {
+            String location = SubService.resolveLocation(ip);
+            if (location != null && !location.isBlank() && !location.equals("unknown")) {
+                try { subscriptions.updateSubUpdateRecordLocation(recordId, location); }
+                catch (Exception ignored) { /* Optional display enrichment; no request failure or sensitive log. */ }
+            }
+        }); } catch (RejectedExecutionException ignored) { }
+    }
+
+    public static String sha256(byte[] body) {
+        try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(body)); }
+        catch (Exception failure) { throw new IllegalStateException("SHA-256 unavailable"); }
+    }
+
+    @PreDestroy void close() { locations.shutdownNow(); }
+}

src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java

@@ -9,6 +9,7 @@ import com.lion.lionwebsite.Util.CustomUtil;
 import com.lion.lionwebsite.Util.SubscriptionCacheStore;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import jakarta.annotation.PostConstruct;
 
@@ -29,6 +30,9 @@ public class SubscriptionStandbySnapshotService {
     private final SubscriptionStateCoordinator stateCoordinator;
     private final ObjectMapper objectMapper = CustomUtil.objectMapper;
 
+    @Autowired(required = false)
+    SubscriptionEventService subscriptionEvents;
+
     @Value("${subscription.standby.sync-secret:}")
     String syncSecret;
 
@@ -78,7 +82,9 @@ public class SubscriptionStandbySnapshotService {
         List<SubscriptionAccountSnapshot> accounts = new ArrayList<>(accountMap.values());
         accounts.sort(Comparator.comparing(SubscriptionAccountSnapshot::getAccountId));
         List<SubscriptionBindingSnapshot> bindings = new ArrayList<>();
-        for (SubBind bind : subMapper.selectAllSubBind()) {
+        List<SubBind> currentBindings = subMapper.selectAllSubBind();
+        if (subscriptionEvents != null) subscriptionEvents.rememberBindings(currentBindings);
+        for (SubBind bind : currentBindings) {
             if (bind.getSubscriptionAccountId() == null || !accountMap.containsKey(bind.getSubscriptionAccountId()))
                 continue;
             SubscriptionBindingSnapshot snapshot = new SubscriptionBindingSnapshot();