package jnpf.message.mapper;

import jnpf.message.entity.MessageReceiveEntity;
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.apache.ibatis.mapping.Environment;
import org.apache.ibatis.session.*;
import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import static org.junit.jupiter.api.Assertions.*;

@EnabledIfEnvironmentVariable(named="DMS_READ_TEST_URL", matches=".+")
class DmsUpgradeMessagePostgresTest {
    @Test void receiptAndInboxRollbackTogetherAndDeduplicateAfterLostAcknowledgement() throws Exception {
        var ds=new UnpooledDataSource("org.postgresql.Driver",System.getenv("DMS_READ_TEST_URL"),System.getenv("DMS_READ_TEST_USER"),System.getenv("DMS_READ_TEST_PASSWORD"));
        var config=new Configuration(new Environment("message-test",new JdbcTransactionFactory(),ds));
        String path="mapper/DmsUpgradeMessageMapper.xml";
        try(var input=getClass().getResourceAsStream("/"+path)) {
            String xml=new String(Objects.requireNonNull(input).readAllBytes(),StandardCharsets.UTF_8).replace("public.","pg_temp.");
            new XMLMapperBuilder(new StringReader(xml),config,path,config.getSqlFragments()).parse();
        }
        try(var session=new SqlSessionFactoryBuilder().build(config).openSession(false)) {
            try {
                var c=session.getConnection();
                try(var sql=c.createStatement()) {
                    sql.execute("CREATE TEMP TABLE base_msg_template (LIKE public.base_msg_template INCLUDING DEFAULTS) ON COMMIT DROP");
                    sql.execute("CREATE TEMP TABLE base_message (LIKE public.base_message INCLUDING DEFAULTS INCLUDING INDEXES) ON COMMIT DROP");
                    sql.execute("CREATE TEMP TABLE dms_template_upgrade_message_receipts (tenant_id text,task_id text,user_id text,created_at timestamp default CURRENT_TIMESTAMP,PRIMARY KEY(tenant_id,task_id,user_id)) ON COMMIT DROP");
                    sql.execute("INSERT INTO pg_temp.base_msg_template(f_id,f_tenant_id,f_en_code,f_message_type,f_message_source,f_enabled_mark,f_title,f_content) VALUES ('template','test','dmsTemplateUpgrade','1','3',1,'title','body')");
                }
                var mapper=session.getMapper(DmsUpgradeMessageMapper.class);
                assertEquals("body",mapper.templates("test").getFirst().getContent()); assertTrue(mapper.templates("other").isEmpty());
                var savepoint=c.setSavepoint();
                assertEquals(1,mapper.claimDelivery("test","task","user"));
                var message=new MessageReceiveEntity(); message.setId("m1");message.setUserId("user");message.setTitle("title");message.setBodyText("body");
                assertEquals(1,mapper.insertMessage(message,"test"));
                assertThrows(Exception.class,()->mapper.insertMessage(message,"test"));
                c.rollback(savepoint);session.clearCache();
                assertEquals(1,mapper.claimDelivery("test","task","user"));
                assertEquals(1,mapper.insertMessage(message,"test"));
                assertEquals(0,mapper.claimDelivery("test","task","user"));
                assertEquals(1,mapper.claimDelivery("test","next-upgrade","user"));
                assertEquals(1,mapper.claimDelivery("other","task","user"));
                try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT count(*) FROM pg_temp.base_message")) { assertTrue(rows.next());assertEquals(1,rows.getInt(1)); }
            } finally { session.rollback(true); }
        }
    }
    @Test void serviceTransactionRollsBackAllRecipientsAndAcknowledgesOnlyCommittedMessages() throws Exception {
        try(var c=java.sql.DriverManager.getConnection(System.getenv("DMS_READ_TEST_URL"),System.getenv("DMS_READ_TEST_USER"),System.getenv("DMS_READ_TEST_PASSWORD"));
            var pushes=org.mockito.Mockito.mockStatic(jnpf.message.util.PushMessageUtil.class)) {
            var ds=new org.springframework.jdbc.datasource.SingleConnectionDataSource(c,true);
            var config=new Configuration(new Environment("message-service-test",new org.mybatis.spring.transaction.SpringManagedTransactionFactory(),ds));
            String path="mapper/DmsUpgradeMessageMapper.xml";
            try(var input=getClass().getResourceAsStream("/"+path)) {
                String xml=new String(Objects.requireNonNull(input).readAllBytes(),StandardCharsets.UTF_8).replace("public.","pg_temp.");
                new XMLMapperBuilder(new StringReader(xml),config,path,config.getSqlFragments()).parse();
            }
            try(var sql=c.createStatement()) {
                sql.execute("CREATE TEMP TABLE base_msg_template (LIKE public.base_msg_template INCLUDING DEFAULTS)");
                sql.execute("CREATE TEMP TABLE base_message (LIKE public.base_message INCLUDING DEFAULTS INCLUDING INDEXES)");
                sql.execute("ALTER TABLE pg_temp.base_message ADD CONSTRAINT test_failure CHECK(f_user_id != 'fail')");
                sql.execute("CREATE TEMP TABLE dms_template_upgrade_message_receipts (tenant_id text,task_id text,user_id text,created_at timestamp default CURRENT_TIMESTAMP,PRIMARY KEY(tenant_id,task_id,user_id))");
                sql.execute("INSERT INTO pg_temp.base_msg_template(f_id,f_tenant_id,f_en_code,f_message_type,f_message_source,f_enabled_mark,f_title,f_content) VALUES ('template','test','dmsTemplateUpgrade','1','3',1,'title','body')");
            }
            var session=new org.mybatis.spring.SqlSessionTemplate(new SqlSessionFactoryBuilder().build(config));
            var target=new jnpf.message.service.DmsUpgradeMessageService(session.getMapper(DmsUpgradeMessageMapper.class));
            var proxy=new org.springframework.aop.framework.ProxyFactory(target);
            proxy.addAdvice(new org.springframework.transaction.interceptor.TransactionInterceptor(
                new org.springframework.jdbc.datasource.DataSourceTransactionManager(ds),new org.springframework.transaction.annotation.AnnotationTransactionAttributeSource()));
            var service=(jnpf.message.service.DmsUpgradeMessageService)proxy.getProxy();
            var request=new jnpf.message.model.DmsUpgradeMessage();request.setTenantId("test");request.setTaskId("task");request.setActorId("actor");
            request.setToUserIds(java.util.List.of("user","fail"));
            assertThrows(Exception.class,()->service.send(request));
            pushes.verifyNoInteractions();
            try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT (SELECT count(*) FROM pg_temp.base_message)+(SELECT count(*) FROM pg_temp.dms_template_upgrade_message_receipts)")) {
                assertTrue(rows.next());assertEquals(0,rows.getInt(1));
            }
            request.setToUserIds(java.util.List.of("user")); assertTrue(service.send(request)); assertTrue(service.send(request));
            pushes.verify(()->jnpf.message.util.PushMessageUtil.pushMessage(org.mockito.ArgumentMatchers.anyMap(),org.mockito.ArgumentMatchers.any(),org.mockito.ArgumentMatchers.eq(3)),org.mockito.Mockito.times(1));
            try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT count(*) FROM pg_temp.base_message")) { assertTrue(rows.next());assertEquals(1,rows.getInt(1)); }
        }
    }

}