package com.baiyue.auth.service;

import com.baiyue.auth.client.AuthorizationLoader;
import com.baiyue.auth.client.AuthorizationSnapshot;
import com.baiyue.auth.config.AuthSecurityProperties;
import com.baiyue.auth.config.UnifiedJwtProperties;
import com.baiyue.auth.dto.AccountManagementDtos.ResetPasswordRequest;
import com.baiyue.auth.dto.LoginRequest;
import com.baiyue.auth.entity.AuthAccount;
import com.baiyue.auth.repository.AuthAccountRepository;
import com.baiyue.common.core.exception.BizException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.security.oauth2.jwt.Jwt;
import org.springframework.security.oauth2.jwt.JwtEncoder;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.Duration;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;

/** Real MyBatis repository, JDBC transactions/row locks and production login/refresh/revocation services.
 * Redis is an atomic in-memory transport double; this is not a Redis/MySQL environment test. */
class SessionStateConcurrencyTest {
    JdbcTemplate authDb, iamDb;
    DataSourceTransactionManager authTx, iamTx;
    AuthService auth;
    AccountManagementService accounts;
    SessionStateService states;
    TokenSessionStore sessions;
    JwtEncoder encoder;
    AtomicBoolean redisFailure;
    final ConcurrentMap<String, String> values = new ConcurrentHashMap<>();
    final ConcurrentMap<String, Set<String>> sets = new ConcurrentHashMap<>();

    @BeforeEach void setup() throws Exception {
        authDb = database(); iamDb = database();
        authTx = new DataSourceTransactionManager(authDb.getDataSource());
        iamTx = new DataSourceTransactionManager(iamDb.getDataSource());
        authDb.execute("CREATE TABLE auth_account(id BIGINT PRIMARY KEY, username VARCHAR(50),phone VARCHAR(50),"
            + "password_hash VARCHAR(100),display_name VARCHAR(50),account_status VARCHAR(20),failed_login_count INT,"
            + "locked_until TIMESTAMP,token_version INT,password_change_required BOOLEAN,default_tenant_id BIGINT,"
            + "operator_tenant_id BIGINT,default_membership_id BIGINT,default_app_code VARCHAR(20),deleted INT,"
            + "last_login_at TIMESTAMP,created_at TIMESTAMP,updated_at TIMESTAMP)");
        authDb.update("INSERT INTO auth_account(id,username,password_hash,account_status,token_version,default_tenant_id,"
            + "operator_tenant_id,default_membership_id,default_app_code,deleted) VALUES(1,'operator','old-password','ACTIVE',1,7,7,9,'MERCHANT',0)");
        iamDb.execute("CREATE TABLE authorization_state(id BIGINT PRIMARY KEY, version BIGINT, tenant_version BIGINT, permission VARCHAR(50), warehouse BIGINT, active BOOLEAN)");
        iamDb.update("INSERT INTO authorization_state VALUES (1,1,1,'write',10,true)");

        var factory = new com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean();
        factory.setDataSource(authDb.getDataSource());
        var config = new com.baomidou.mybatisplus.core.MybatisConfiguration();
        config.setMapUnderscoreToCamelCase(true);
        config.addMapper(com.baiyue.auth.mapper.AuthAccountMapper.class);
        factory.setConfiguration(config);
        var mapper = new org.mybatis.spring.SqlSessionTemplate(factory.getObject())
            .getMapper(com.baiyue.auth.mapper.AuthAccountMapper.class);
        var repository = new AuthAccountRepository();
        ReflectionTestUtils.setField(repository, "mapper", mapper);
        AuthorizationLoader loader = mock(AuthorizationLoader.class);
        when(loader.load(any())).thenAnswer(call -> {
            var tx = new TransactionTemplate(iamTx);
            tx.setIsolationLevel(org.springframework.transaction.TransactionDefinition.ISOLATION_REPEATABLE_READ);
            return tx.execute(status -> iamDb.queryForObject("SELECT * FROM authorization_state WHERE id=1", (rs, row) -> {
                if (!rs.getBoolean("active")) throw new IllegalArgumentException("IAM identity disabled");
                return new AuthorizationSnapshot(1L, 7L, 9L, "MERCHANT", "MERCHANT", Set.of("OPERATOR"),
                    Set.of(rs.getString("permission")), rs.getLong("version"), rs.getLong("warehouse"), null,
                    rs.getLong("tenant_version"));
            }));
        });
        sessions = new TokenSessionStore();
        ReflectionTestUtils.setField(sessions, "objectMapper", new ObjectMapper());
        ReflectionTestUtils.setField(sessions, "keyPrefix", "test");
        ReflectionTestUtils.setField(sessions, "redisTemplate", redis());
        PasswordEncoder passwords = mock(PasswordEncoder.class);
        when(passwords.matches(anyString(), anyString())).thenAnswer(c -> c.getArgument(0).equals(c.getArgument(1)));
        when(passwords.encode(anyString())).thenAnswer(c -> c.getArgument(0));
        encoder = mock(JwtEncoder.class);
        when(encoder.encode(any())).thenAnswer(c -> {
            var claims = ((org.springframework.security.oauth2.jwt.JwtEncoderParameters)c.getArgument(0)).getClaims();
            return Jwt.withTokenValue(claims.getId()).header("alg", "RS256").claims(m -> m.putAll(claims.getClaims())).build();
        });
        var service = new AuthService();
        ReflectionTestUtils.setField(service, "accountRepository", repository);
        ReflectionTestUtils.setField(service, "authorizationLoader", loader);
        ReflectionTestUtils.setField(service, "tokenSessionStore", sessions);
        ReflectionTestUtils.setField(service, "passwordEncoder", passwords);
        ReflectionTestUtils.setField(service, "jwtEncoder", encoder);
        ReflectionTestUtils.setField(service, "properties", new AuthSecurityProperties("issuer", "aud", Duration.ofMinutes(5), Duration.ofHours(1), 5, Duration.ofMinutes(1)));
        ReflectionTestUtils.setField(service, "signingProperties", new UnifiedJwtProperties("unused", "kid"));
        auth = transactional(service);
        var management = new AccountManagementService();
        ReflectionTestUtils.setField(management, "repository", repository);
        ReflectionTestUtils.setField(management, "tokenSessionStore", sessions);
        ReflectionTestUtils.setField(management, "passwordEncoder", passwords);
        accounts = transactional(management);
        var state = new SessionStateService();
        ReflectionTestUtils.setField(state, "accounts", repository);
        ReflectionTestUtils.setField(state, "authorizationLoader", loader);
        ReflectionTestUtils.setField(state, "sessions", sessions);
        states = transactional(state);
    }

    @Test void lateLoginAfterRevokeBeforeIamCommitCannotUseOldPermissionsOrRefresh() throws Exception {
        var revoked = new CountDownLatch(1);
        var snapshotLoaded = new CountDownLatch(1);
        var committed = new CountDownLatch(1);
        doAnswer(c -> {
            snapshotLoaded.countDown(); await(committed);
            var claims = ((org.springframework.security.oauth2.jwt.JwtEncoderParameters)c.getArgument(0)).getClaims();
            return Jwt.withTokenValue(claims.getId()).header("alg", "RS256").claims(m -> m.putAll(claims.getClaims())).build();
        }).when(encoder).encode(any());
        ExecutorService pool = Executors.newFixedThreadPool(2);
        try {
            Future<?> change = pool.submit(() -> {
                new TransactionTemplate(iamTx).executeWithoutResult(tx -> {
                    iamDb.update("UPDATE authorization_state SET version=version+1, permission='read' WHERE id=1");
                    accounts.revokeSessions(Set.of(1L));
                    revoked.countDown(); await(snapshotLoaded);
                });
                committed.countDown();
            });
            await(revoked);
            var login = pool.submit(() -> auth.login(new LoginRequest("operator", "old-password")));
            var stale = login.get(5, TimeUnit.SECONDS); change.get(5, TimeUnit.SECONDS);
            // Access and state were written AFTER revoke, yet the first post-commit request is denied.
            assertThat(values).containsKey("test:access:" + stale.accessToken());
            assertThatThrownBy(() -> states.validate(stale.accessToken(), 1L)).isInstanceOf(BizException.class);
            assertThatThrownBy(() -> auth.refresh(stale.refreshToken())).isInstanceOf(BizException.class);
            var fresh = auth.login(new LoginRequest("operator", "old-password"));
            assertThat(states.validate(fresh.accessToken(), 1L)).isTrue();
            assertThat(fresh.authorities()).containsExactly("read");
            assertThatThrownBy(() -> auth.refresh(stale.refreshToken())).isInstanceOf(BizException.class);
        } finally { committed.countDown(); pool.shutdownNow(); }
    }

    @Test void passwordResetWaitsForConcurrentLoginThenInvalidatesAccessAndRefresh() throws Exception {
        var loaded = new CountDownLatch(1); var release = new CountDownLatch(1);
        doAnswer(c -> {
            loaded.countDown(); await(release);
            var claims = ((org.springframework.security.oauth2.jwt.JwtEncoderParameters)c.getArgument(0)).getClaims();
            return Jwt.withTokenValue(claims.getId()).header("alg", "RS256").claims(m -> m.putAll(claims.getClaims())).build();
        }).when(encoder).encode(any());
        ExecutorService pool = Executors.newFixedThreadPool(2);
        try {
            var login = pool.submit(() -> auth.login(new LoginRequest("operator", "old-password")));
            await(loaded);
            var resetting = new CountDownLatch(1);
            var reset = pool.submit(() -> { resetting.countDown(); accounts.resetPassword(1L, new ResetPasswordRequest("new-password", false), false); });
            await(resetting);
            assertThatThrownBy(() -> reset.get(100, TimeUnit.MILLISECONDS)).isInstanceOf(TimeoutException.class);
            release.countDown();
            var old = login.get(5, TimeUnit.SECONDS); reset.get(5, TimeUnit.SECONDS);
            assertThat(authDb.queryForObject("SELECT token_version FROM auth_account WHERE id=1", Integer.class)).isEqualTo(2);
            assertThatThrownBy(() -> states.validate(old.accessToken(), 1L)).isInstanceOf(BizException.class);
            assertThatThrownBy(() -> auth.refresh(old.refreshToken())).isInstanceOf(BizException.class);
        } finally { release.countDown(); pool.shutdownNow(); }
    }

    @Test void failedRevocationRollsBackAuthAndIamAndDoesNotReportSuccess() {
        var old = auth.login(new LoginRequest("operator", "old-password"));
        redisFailure.set(true);
        assertThatThrownBy(() -> new TransactionTemplate(iamTx).executeWithoutResult(tx -> {
            iamDb.update("UPDATE authorization_state SET version=version+1, active=false WHERE id=1");
            accounts.revokeSessions(Set.of(1L));
        })).isInstanceOf(BizException.class);
        redisFailure.set(false);
        assertThat(authDb.queryForObject("SELECT token_version FROM auth_account WHERE id=1", Integer.class)).isEqualTo(1);
        assertThat(iamDb.queryForObject("SELECT version FROM authorization_state WHERE id=1", Long.class)).isEqualTo(1);
        assertThat(states.validate(old.accessToken(), 1L)).isTrue();
    }

    @Test void committedIamVersionRejectsResidualSessionWhenNoRedisDeletionHappens() {
        var old = auth.login(new LoginRequest("operator", "old-password"));
        new TransactionTemplate(iamTx).executeWithoutResult(tx ->
            iamDb.update("UPDATE authorization_state SET tenant_version=tenant_version+1 WHERE id=1"));
        assertThatThrownBy(() -> states.validate(old.accessToken(), 1L)).isInstanceOf(BizException.class);
        assertThatThrownBy(() -> auth.refresh(old.refreshToken())).isInstanceOf(BizException.class);
        var current = auth.login(new LoginRequest("operator", "old-password"));
        assertThat(states.validate(current.accessToken(), 1L)).isTrue();
        iamDb.update("UPDATE authorization_state SET warehouse=20 WHERE id=1");
        assertThatThrownBy(() -> states.validate(current.accessToken(), 1L)).isInstanceOf(BizException.class);
    }

    @Test void iamRollbackPreservesFingerprintAndOtherUserCannotUseSession() {
        var old = auth.login(new LoginRequest("operator", "old-password"));
        new TransactionTemplate(iamTx).executeWithoutResult(tx -> {
            iamDb.update("UPDATE authorization_state SET version=version+1, permission='read' WHERE id=1");
            tx.setRollbackOnly();
        });
        assertThat(states.validate(old.accessToken(), 1L)).isTrue();
        assertThatThrownBy(() -> states.validate(old.accessToken(), 2L)).isInstanceOf(BizException.class);
        var refreshed = auth.refresh(old.refreshToken());
        assertThat(states.validate(refreshed.accessToken(), 1L)).isTrue();
    }

    @Test void lostRevokeAcknowledgementRollsBackIamButCannotRestoreOldAuthSession() {
        var old = auth.login(new LoginRequest("operator", "old-password"));
        assertThatThrownBy(() -> new TransactionTemplate(iamTx).executeWithoutResult(tx -> {
            iamDb.update("UPDATE authorization_state SET version=version+1, permission='read' WHERE id=1");
            accounts.revokeSessions(Set.of(1L));
            throw new IllegalStateException("Auth response timed out after its commit");
        })).isInstanceOf(IllegalStateException.class);
        assertThat(iamDb.queryForObject("SELECT permission FROM authorization_state WHERE id=1", String.class)).isEqualTo("write");
        assertThat(authDb.queryForObject("SELECT token_version FROM auth_account WHERE id=1", Integer.class)).isEqualTo(2);
        assertThatThrownBy(() -> states.validate(old.accessToken(), 1L)).isInstanceOf(BizException.class);
        var fresh = auth.login(new LoginRequest("operator", "old-password"));
        assertThat(states.validate(fresh.accessToken(), 1L)).isTrue();
        assertThat(fresh.authorities()).containsExactly("write");
    }

    @Test void disabledCustomerOrMemberAndUnavailableIamRejectAccessAndRefresh() {
        var old = auth.login(new LoginRequest("operator", "old-password"));
        // No Redis deletion: the authoritative identity gate alone closes the session.
        iamDb.update("UPDATE authorization_state SET active=false, tenant_version=tenant_version+1 WHERE id=1");
        assertThatThrownBy(() -> states.validate(old.accessToken(), 1L)).isInstanceOf(IllegalArgumentException.class);
        assertThatThrownBy(() -> auth.refresh(old.refreshToken())).isInstanceOf(IllegalArgumentException.class);
        iamDb.update("UPDATE authorization_state SET active=true WHERE id=1");
        var fresh = auth.login(new LoginRequest("operator", "old-password"));
        iamDb.execute("DROP TABLE authorization_state");
        assertThatThrownBy(() -> states.validate(fresh.accessToken(), 1L)).isInstanceOf(org.springframework.dao.DataAccessException.class);
        assertThatThrownBy(() -> auth.refresh(fresh.refreshToken())).isInstanceOf(org.springframework.dao.DataAccessException.class);
    }

    @SuppressWarnings("unchecked") private StringRedisTemplate redis() {
        redisFailure = new AtomicBoolean();
        var redis = mock(StringRedisTemplate.class);
        ValueOperations<String,String> ops = mock(ValueOperations.class);
        SetOperations<String,String> index = mock(SetOperations.class);
        when(redis.opsForValue()).thenReturn(ops); when(redis.opsForSet()).thenReturn(index);
        doAnswer(c -> { values.put(c.getArgument(0), c.getArgument(1)); return null; }).when(ops).set(anyString(), anyString(), any(Duration.class));
        when(ops.get(anyString())).thenAnswer(c -> values.get(c.getArgument(0)));
        when(ops.getAndDelete(anyString())).thenAnswer(c -> values.remove(c.getArgument(0)));
        when(redis.hasKey(anyString())).thenAnswer(c -> values.containsKey(c.getArgument(0)));
        when(redis.delete(anyString())).thenAnswer(c -> values.remove(c.getArgument(0)) != null);
        when(redis.delete(anyCollection())).thenAnswer(c -> {
            if (redisFailure.get()) throw new org.springframework.data.redis.RedisConnectionFailureException("offline");
            long count = 0; for (Object key : (java.util.Collection<?>) c.getArgument(0)) if (values.remove(key) != null) count++;
            return count;
        });
        when(index.members(anyString())).thenAnswer(c -> sets.getOrDefault(c.getArgument(0), Set.of()));
        doAnswer(c -> {
            Set<String> members = sets.computeIfAbsent(c.getArgument(0), key -> ConcurrentHashMap.newKeySet());
            for (int i=1; i<c.getArguments().length; i++) members.add(c.getArgument(i));
            return 1L;
        }).when(index).add(anyString(), any(String[].class));
        return redis;
    }

    static JdbcTemplate database() {
        var ds = new JdbcDataSource(); ds.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=3000");
        return new JdbcTemplate(ds);
    }
    @SuppressWarnings("unchecked") <T> T transactional(T target) {
        var factory = new ProxyFactory(target); factory.setProxyTargetClass(true);
        factory.addAdvice(new TransactionInterceptor(authTx, new AnnotationTransactionAttributeSource()));
        return (T) factory.getProxy();
    }
    static void await(CountDownLatch latch) {
        try { if (!latch.await(5, TimeUnit.SECONDS)) throw new AssertionError("Barrier timed out"); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new AssertionError(e); }
    }
}