package study.u60;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.List;
import java.util.UUID;

/**
 * U60 真实 Redis（租约授权权威）+ 真实 MySQL（资源条件更新）验证租约/fencing 协议。
 * 权威与资源是两个独立服务、两个独立持久状态，二者不共享事务——正是协议的前提。
 */
public class LeaseFencingLab {

    static final String RESOURCE_ID = "u60-resource-1";

    public static void main(String[] args) throws Exception {
        final String schema = args[0];
        final int redisPort = Integer.parseInt(args[1]);
        final String mysqlUrl = "jdbc:mysql://127.0.0.1:3306/" + schema + "?useSSL=false&allowPublicKeyRetrieval=true";

        check(schema.startsWith("study_u60_"), "schema 前缀必须为 study_u60_");
        try (Connection probe = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/", "root", "");
             Statement st = probe.createStatement()) {
            st.execute("CREATE DATABASE IF NOT EXISTS " + schema);
            try (ResultSet rs = st.executeQuery("SELECT VERSION()")) {
                rs.next();
                out("mysql=" + rs.getString(1));
            }
        }
        out("schema-created=" + schema);
        JedisPool redis = new JedisPool("127.0.0.1", redisPort);
        try (Jedis j = redis.getResource()) {
            String info = j.info("server");
            String version = "unknown";
            for (String line : info.split("\r?\n")) {
                if (line.startsWith("redis_version:")) version = line.substring("redis_version:".length()).trim();
            }
            out("redis=" + version);
        }

        try {
            try (Connection c = DriverManager.getConnection(mysqlUrl, "root", "")) {
                try (Statement st = c.createStatement()) {
                    st.execute("CREATE TABLE resource (" +
                            "id VARCHAR(64) PRIMARY KEY," +
                            "value VARCHAR(128) NOT NULL," +
                            "max_fence BIGINT NOT NULL DEFAULT 0," +
                            "last_seq BIGINT NOT NULL DEFAULT 0," +
                            "last_owner VARCHAR(64) NOT NULL)");
                    st.execute("INSERT INTO resource (id, value, max_fence, last_seq, last_owner)" +
                            " VALUES ('" + RESOURCE_ID + "', 'initial', 0, 0, 'none')");
                }
                int tables = 0;
                try (PreparedStatement ps = c.prepareStatement(
                        "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=? AND table_name='resource'")) {
                    ps.setString(1, schema);
                    try (ResultSet rs = ps.executeQuery()) {
                        rs.next();
                        tables = rs.getInt(1);
                    }
                }
                check(tables == 1, "资源表应位于独有 schema");
                out("target=PASS wrong-schema=rejected actual-schema=verified");

                // ---- 1. 租约获取：generation 递增、owner/expiry/request_id 与授予同属一个 Lua 原子权威 ----
                String grantScript = """
                        local gen = redis.call('INCR', KEYS[3])
                        redis.call('HSET', KEYS[1], 'owner', ARGV[1], 'request_id', ARGV[3], 'generation', gen)
                        redis.call('PEXPIRE', KEYS[1], ARGV[2])
                        redis.call('SADD', KEYS[2], ARGV[3])
                        return gen""";
                String ownerA = "A-" + UUID.randomUUID().toString().substring(0, 8);
                String requestIdA = UUID.randomUUID().toString();
                long fenceA;
                try (Jedis j = redis.getResource()) {
                    Object gen = j.eval(grantScript, List.of("u60:lease:res1", "u60:grants", "u60:gen:res1"),
                            List.of(ownerA, "800", requestIdA));
                    fenceA = ((Number) gen).longValue();
                    long px = j.pttl("u60:lease:res1");
                    check(fenceA == 1 && px > 0, "首次授予应得 fence=1 且租期生效");
                }
                try (Jedis j = redis.getResource()) {
                    check(j.sismember("u60:grants", requestIdA), "授予的 request_id 幂等记录应在权威集合中");
                }
                out("leaseAcquire fence=1 generationAtomic=true requestRecorded=true");

                // ---- 2. owner 校验续租：非持有者不能续租 ----
                String renewScript = """
                        if redis.call('HGET', KEYS[1], 'owner') == ARGV[1] then
                          redis.call('PEXPIRE', KEYS[1], ARGV[2])
                          return 1
                        else
                          return -1
                        end""";
                try (Jedis j = redis.getResource()) {
                    Object r = j.eval(renewScript, List.of("u60:lease:res1"), List.of("B-impostor", "800"));
                    check(((Number) r).longValue() == -1, "非持有者续租应被拒绝");
                    Object ok = j.eval(renewScript, List.of("u60:lease:res1"), List.of(ownerA, "800"));
                    check(((Number) ok).longValue() == 1, "持有者续租应成功");
                }
                out("leaseRenew ownerChecked=true impostorRejected=true");

                // ---- 3. A 过期：租约到期后 B 获更高 fence ----
                Thread.sleep(900); // PX=800ms 过期；暂停用真实等待表示，不用时钟推断
                String ownerB = "B-" + UUID.randomUUID().toString().substring(0, 8);
                long fenceB;
                try (Jedis j = redis.getResource()) {
                    Object gen = j.eval(grantScript, List.of("u60:lease:res1", "u60:grants", "u60:gen:res1"),
                            List.of(ownerB, "800", UUID.randomUUID().toString()));
                    fenceB = ((Number) gen).longValue();
                    check(fenceB == 2, "B 应获得递增 generation=2，实际 " + fenceB);
                }
                out("leaseExpiry takeover=true fenceA=1 fenceB=2");

                // ---- 4. B 以 fence=2 提交资源；随后 A 以过期 fence=1 提交被拒（41/42/41 时间线） ----
                long updatedB = conditionalWrite(mysqlUrl, RESOURCE_ID, "state-by-B", fenceB, 1, ownerB);
                check(updatedB == 1, "新代际 B(fence=2,seq=1) 应写入成功");
                long updatedA = conditionalWrite(mysqlUrl, RESOURCE_ID, "stale-by-A", fenceA, 1, ownerA);
                check(updatedA == 0, "旧代际 A(fence=1) 必须被资源拒绝，实际影响行数 " + updatedA);
                check("state-by-B".equals(readValue(mysqlUrl, RESOURCE_ID)), "资源持久值应保持 B 的状态");
                out("fencing staleRejected=true persistedValue=state-by-B maxFence=2");

                // ---- 5. 同任期 op_seq 顺序：seq2 先到、seq1 后到应被拒 ----
                long seq2 = conditionalWrite(mysqlUrl, RESOURCE_ID, "op-2", fenceB, 2, ownerB);
                long seq1 = conditionalWrite(mysqlUrl, RESOURCE_ID, "op-1", fenceB, 1, ownerB);
                check(seq2 == 1 && seq1 == 0, "同任期 (fence,seq) 字典序应拒绝迟到 seq1");
                check("op-2".equals(readValue(mysqlUrl, RESOURCE_ID)), "资源值应为 seq2 的状态");
                out("opSeq ordering=lexicographic lateSeqRejected=true");

                // ---- 6. 业务幂等：同键同摘要返回原结果，同键异参报冲突 ----
                long dupSame = conditionalWriteIfAbsent(mysqlUrl, RESOURCE_ID, fenceB, 2, "op-2");
                check(dupSame == -1, "同键同摘要应返回原结果（不产生新副作用）");
                long dupDiff = conditionalWriteIfAbsent(mysqlUrl, RESOURCE_ID, fenceB, 2, "different");
                check(dupDiff == -2, "同键异参应报冲突");
                out("idempotency sameKeyReturnsOriginal=true differentArgsConflict=true");
            }
        } finally {
            try (Jedis j = redis.getResource()) {
                j.del("u60:lease:res1", "u60:grants", "u60:gen:res1");
            }
            out("cleanup redisKeys=removed");
            redis.close();
            try (Connection cleanup = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/", "root", "");
                 Statement st = cleanup.createStatement()) {
                st.execute("DROP DATABASE IF EXISTS " + schema);
                try (ResultSet rs = st.executeQuery("SHOW DATABASES LIKE '" + schema + "'")) {
                    if (!rs.next()) out("cleanup own-schema=removed");
                }
            }
        }
    }

    /** 资源条件更新：原子检查 (max_fence,last_seq) 字典序 + 业务写，同一事务。 */
    private static long conditionalWrite(String mysqlUrl, String id, String value,
                                         long fence, long seq, String owner) throws Exception {
        try (Connection conn = DriverManager.getConnection(mysqlUrl, "root", "")) {
            conn.setAutoCommit(false);
            try (PreparedStatement ps = conn.prepareStatement(
                    "UPDATE resource SET value=?, max_fence=?, last_seq=?, last_owner=?" +
                            " WHERE id=? AND (max_fence < ? OR (max_fence = ? AND last_seq < ?))")) {
                ps.setString(1, value);
                ps.setLong(2, fence);
                ps.setLong(3, seq);
                ps.setString(4, owner);
                ps.setString(5, id);
                ps.setLong(6, fence);
                ps.setLong(7, fence);
                ps.setLong(8, seq);
                int rows = ps.executeUpdate();
                conn.commit();
                return rows;
            } catch (Exception e) {
                conn.rollback();
                throw e;
            } finally {
                conn.setAutoCommit(true);
            }
        }
    }

    /** 带业务身份的提交：同键同摘要返回 -1（原结果），同键异参返回 -2（冲突）。 */
    private static long conditionalWriteIfAbsent(String mysqlUrl, String id,
                                                 long fence, long seq, String newValue) throws Exception {
        try (Connection conn = DriverManager.getConnection(mysqlUrl, "root", "")) {
            conn.setAutoCommit(false);
            try (PreparedStatement q = conn.prepareStatement(
                    "SELECT value FROM resource WHERE id=? AND max_fence=? AND last_seq=?")) {
                q.setString(1, id);
                q.setLong(2, fence);
                q.setLong(3, seq);
                try (ResultSet rs = q.executeQuery()) {
                    if (rs.next()) {
                        conn.rollback();
                        return newValue.equals(rs.getString(1)) ? -1 : -2;
                    }
                }
            }
            try (PreparedStatement ps = conn.prepareStatement(
                    "UPDATE resource SET value=?, max_fence=?, last_seq=?, last_owner=?" +
                            " WHERE id=? AND (max_fence < ? OR (max_fence = ? AND last_seq < ?))")) {
                ps.setString(1, newValue);
                ps.setLong(2, fence);
                ps.setLong(3, seq);
                ps.setString(4, "retry-" + newValue);
                ps.setString(5, id);
                ps.setLong(6, fence);
                ps.setLong(7, fence);
                ps.setLong(8, seq);
                int rows = ps.executeUpdate();
                conn.commit();
                return rows;
            } catch (Exception e) {
                conn.rollback();
                throw e;
            } finally {
                conn.setAutoCommit(true);
            }
        }
    }

    private static String readValue(String mysqlUrl, String id) throws Exception {
        try (Connection conn = DriverManager.getConnection(mysqlUrl, "root", "");
             PreparedStatement ps = conn.prepareStatement("SELECT value FROM resource WHERE id=?")) {
            ps.setString(1, id);
            try (ResultSet rs = ps.executeQuery()) {
                rs.next();
                return rs.getString(1);
            }
        }
    }

    static void check(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
    }

    static void out(String line) {
        System.out.println("U60 " + line);
    }
}