package jnpf.base.service;

import com.baomidou.dynamic.datasource.tx.TransactionContext;
import jnpf.base.ActionResult;
import jnpf.exception.DataException;
import jnpf.util.JsonUtil;
import jnpf.util.TicketUtil;
import jnpf.util.UserProvider;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.web.context.request.async.DeferredResult;
import java.util.*;

/** Authorized, content-free change signals. Polling remains authoritative across nodes/restarts. */
@Component
public class UploadEditNotifications {
    private final Map<String, Long> versions = new LinkedHashMap<>();
    private final Set<Waiter> waiters = new HashSet<>();
    private long sequence = System.currentTimeMillis();
    private record Waiter(Map<String, Long> versions, DeferredResult<ActionResult<Boolean>> result) { }

    public synchronized long version(String tenant, String draft) {
        return versions.getOrDefault(tenant + ":" + draft, 0L);
    }

    public String ticket(Map<String, Long> observed) {
        Map<String, Object> proof = new LinkedHashMap<>();
        proof.put("purpose", "upload-edit-change-watch");
        proof.put("tenant", UploadEditDraftService.tenant(UserProvider.getUser()));
        proof.put("user", UserProvider.getLoginUserId());
        proof.put("expires", System.currentTimeMillis() + 60_000L);
        proof.put("versions", observed);
        return TicketUtil.createTicket(JsonUtil.getObjectToString(proof), 60L);
    }

    public DeferredResult<ActionResult<Boolean>> watch(String ticket) {
        String raw = ticket == null ? null : TicketUtil.parseTicket(ticket);
        Map<?, ?> proof = raw == null ? Map.of() : JsonUtil.getJsonToBean(raw, Map.class);
        String tenant = UploadEditDraftService.tenant(UserProvider.getUser());
        if (!"upload-edit-change-watch".equals(proof.get("purpose")) || !tenant.equals(proof.get("tenant"))
                || !Objects.equals(UserProvider.getLoginUserId(), proof.get("user"))
                || !(proof.get("expires") instanceof Number expiry) || expiry.longValue() < System.currentTimeMillis()
                || !(proof.get("versions") instanceof Map<?, ?> observed) || observed.isEmpty() || observed.size() > 200)
            throw new DataException("附件同步凭据已失效，请重新读取共享状态");
        Map<String, Long> expected = new LinkedHashMap<>();
        for (var entry : observed.entrySet()) {
            if (!(entry.getValue() instanceof Number number)) throw new DataException("附件同步凭据无效");
            expected.put(tenant + ":" + entry.getKey(), number.longValue());
        }
        DeferredResult<ActionResult<Boolean>> result = new DeferredResult<>(25_000L, ActionResult.success(false));
        Waiter waiter = new Waiter(expected, result);
        synchronized (this) {
            if (changed(expected)) result.setResult(ActionResult.success(true));
            else if (waiters.size() < 1024) waiters.add(waiter);
            else result.setResult(ActionResult.success(false));
        }
        result.onCompletion(() -> { synchronized (this) { waiters.remove(waiter); } });
        return result;
    }

    public void changedAfterCommit(String tenant, String draft) {
        Runnable publish = () -> publish(tenant + ":" + draft);
        TransactionSynchronization sync = new TransactionSynchronization() {
            @Override public void afterCommit() { publish.run(); }
        };
        if (TransactionContext.getXID() != null) TransactionContext.registerSynchronization(sync);
        else if (TransactionSynchronizationManager.isSynchronizationActive()) TransactionSynchronizationManager.registerSynchronization(sync);
        else publish.run();
    }

    private synchronized boolean changed(Map<String, Long> expected) {
        return expected.entrySet().stream().anyMatch(e -> !Objects.equals(versions.getOrDefault(e.getKey(), 0L), e.getValue()));
    }
    private synchronized void publish(String key) {
        versions.put(key, ++sequence);
        if (versions.size() > 4096) versions.remove(versions.keySet().iterator().next());
        // Completion callbacks may run immediately; iterate a snapshot.
        for (Waiter waiter : new ArrayList<>(waiters)) if (changed(waiter.versions())) {
            waiters.remove(waiter);
            waiter.result().setResult(ActionResult.success(true));
        }
    }
}