<commit_before>package stroom.processor.impl.db;

import stroom.db.util.ExpressionMapper;
import stroom.db.util.ExpressionMapperFactory;
import stroom.db.util.GenericDao;
import stroom.db.util.JooqUtil;
import stroom.entity.shared.ExpressionCriteria;
import stroom.processor.impl.ProcessorDao;
import stroom.processor.impl.db.jooq.tables.records.ProcessorRecord;
import stroom.processor.shared.Processor;
import stroom.processor.shared.ProcessorFields;
import stroom.processor.shared.TaskStatus;
import stroom.util.shared.ResultPage;

import org.jooq.Condition;
import org.jooq.impl.DSL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import javax.inject.Inject;

import static stroom.processor.impl.db.jooq.Tables.PROCESSOR_TASK;
import static stroom.processor.impl.db.jooq.tables.Processor.PROCESSOR;
import static stroom.processor.impl.db.jooq.tables.ProcessorFilter.PROCESSOR_FILTER;

class ProcessorDaoImpl implements ProcessorDao {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProcessorDaoImpl.class);

    private final ProcessorDbConnProvider processorDbConnProvider;
    private final GenericDao<ProcessorRecord, Processor, Integer> genericDao;
    private final ExpressionMapper expressionMapper;

    @Inject
    public ProcessorDaoImpl(final ProcessorDbConnProvider processorDbConnProvider,
                            final ExpressionMapperFactory expressionMapperFactory) {
        this.processorDbConnProvider = processorDbConnProvider;
        this.genericDao = new GenericDao<>(processorDbConnProvider, PROCESSOR, PROCESSOR.ID, Processor.class);

        expressionMapper = expressionMapperFactory.create();
        expressionMapper.map(ProcessorFields.ID, PROCESSOR.ID, Integer::valueOf);
        expressionMapper.map(ProcessorFields.CREATE_USER, PROCESSOR_FILTER.CREATE_USER, value -> value);
        expressionMapper.map(ProcessorFields.PIPELINE, PROCESSOR.PIPELINE_UUID, value -> value);
        expressionMapper.map(ProcessorFields.ENABLED, PROCESSOR.ENABLED, Boolean::valueOf);
        expressionMapper.map(ProcessorFields.DELETED, PROCESSOR.DELETED, Boolean::valueOf);
        expressionMapper.map(ProcessorFields.UUID, PROCESSOR.UUID, value -> value);
    }

    @Override
    public Processor create(final Processor processor) {
        // We don't use the delegate DAO here as we want to handle potential duplicates carefully so this
        // behaves as a getOrCreate method.
        // TODO This should be replaced with JooqUtil.tryCreate as ANY error in the sql will be
        //  ignored, not just key duplicates.
        return JooqUtil.contextResult(processorDbConnProvider, context -> {
            final Optional<ProcessorRecord> optional = context
                    .insertInto(PROCESSOR,
                            PROCESSOR.CREATE_TIME_MS,
                            PROCESSOR.CREATE_USER,
                            PROCESSOR.UPDATE_TIME_MS,
                            PROCESSOR.UPDATE_USER,
                            PROCESSOR.UUID,
                            PROCESSOR.TASK_TYPE,
                            PROCESSOR.PIPELINE_UUID,
                            PROCESSOR.ENABLED,
                            PROCESSOR.DELETED)
                    .values(processor.getCreateTimeMs(),
                            processor.getCreateUser(),
                            processor.getUpdateTimeMs(),
                            processor.getUpdateUser(),
                            processor.getUuid(),
                            processor.getTaskType(),
                            processor.getPipelineUuid(),
                            processor.isEnabled(),
                            processor.isDeleted())
                    .onDuplicateKeyIgnore()
                    .returning(PROCESSOR.ID)
                    .fetchOptional();

            if (optional.isPresent()) {
                final Integer id = optional.get().getId();
                return context
                        .select()
                        .from(PROCESSOR)
                        .where(PROCESSOR.ID.eq(id))
                        .fetchOneInto(Processor.class);
            }

            return context
                    .select()
                    .from(PROCESSOR)
                    .where(PROCESSOR.PIPELINE_UUID.eq(processor.getPipelineUuid()))
                    .fetchOneInto(Processor.class);
        });
    }

    @Override
    public Processor update(final Processor processor) {
        return genericDao.update(processor);
    }

    @Override
    public boolean delete(final int id) {
        return genericDao.delete(id);
    }

    @Override
    public boolean logicalDelete(final int id) {
        try {
            return JooqUtil.transactionResult(processorDbConnProvider, context -> {
                // Logically delete all the child filters first

                final int processor_task_update_count = context
                        .update(PROCESSOR_TASK)
                        .set(PROCESSOR_TASK.STATUS, TaskStatus.DELETED.getPrimitiveValue())
                        .where(DSL.exists(
                                DSL.selectZero()
                                        .from(PROCESSOR_FILTER)
                                        .innerJoin(PROCESSOR)
                                        .on(PROCESSOR_FILTER.FK_PROCESSOR_ID.eq(PROCESSOR.ID))
                                        .where(PROCESSOR.ID.eq(id))))
                        .and(PROCESSOR_TASK.STATUS.eq(TaskStatus.UNPROCESSED.getPrimitiveValue()))
                        .execute();

                LOGGER.debug("Logically deleted {} tasks for processor Id {}",
                        processor_task_update_count, id);

                final int processor_filter_update_count = context
                        .update(PROCESSOR_FILTER)
                        .set(PROCESSOR_FILTER.DELETED, true)
                        .where(PROCESSOR_FILTER.FK_PROCESSOR_ID.eq(id))
                        .execute();
                LOGGER.debug("Logically deleted {} filters for processor Id {}",
                        processor_filter_update_count, id);

                final int processor_update_count = context
                        .update(PROCESSOR)
                        .set(PROCESSOR.DELETED, true)
                        .where(PROCESSOR.ID.eq(id))
                        .execute();

                LOGGER.debug("Logically deleted {} processor with Id {}",
                        processor_update_count, id);

                return processor_update_count > 0;
            });
        } catch (Exception e) {
            throw new RuntimeException(
                    "Error deleting filters, tasks and processor for processor id " + id, e);
        }
    }

    @Override<commit_msg>gh-2889 Add updating of the version col<commit_after>package stroom.processor.impl.db;

import stroom.db.util.ExpressionMapper;
import stroom.db.util.ExpressionMapperFactory;
import stroom.db.util.GenericDao;
import stroom.db.util.JooqUtil;
import stroom.entity.shared.ExpressionCriteria;
import stroom.processor.impl.ProcessorDao;
import stroom.processor.impl.db.jooq.tables.records.ProcessorRecord;
import stroom.processor.shared.Processor;
import stroom.processor.shared.ProcessorFields;
import stroom.processor.shared.TaskStatus;
import stroom.util.shared.ResultPage;

import org.jooq.Condition;
import org.jooq.impl.DSL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import javax.inject.Inject;

import static stroom.processor.impl.db.jooq.Tables.PROCESSOR_TASK;
import static stroom.processor.impl.db.jooq.tables.Processor.PROCESSOR;
import static stroom.processor.impl.db.jooq.tables.ProcessorFilter.PROCESSOR_FILTER;

class ProcessorDaoImpl implements ProcessorDao {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProcessorDaoImpl.class);

    private final ProcessorDbConnProvider processorDbConnProvider;
    private final GenericDao<ProcessorRecord, Processor, Integer> genericDao;
    private final ExpressionMapper expressionMapper;

    @Inject
    public ProcessorDaoImpl(final ProcessorDbConnProvider processorDbConnProvider,
                            final ExpressionMapperFactory expressionMapperFactory) {
        this.processorDbConnProvider = processorDbConnProvider;
        this.genericDao = new GenericDao<>(processorDbConnProvider, PROCESSOR, PROCESSOR.ID, Processor.class);

        expressionMapper = expressionMapperFactory.create();
        expressionMapper.map(ProcessorFields.ID, PROCESSOR.ID, Integer::valueOf);
        expressionMapper.map(ProcessorFields.CREATE_USER, PROCESSOR_FILTER.CREATE_USER, value -> value);
        expressionMapper.map(ProcessorFields.PIPELINE, PROCESSOR.PIPELINE_UUID, value -> value);
        expressionMapper.map(ProcessorFields.ENABLED, PROCESSOR.ENABLED, Boolean::valueOf);
        expressionMapper.map(ProcessorFields.DELETED, PROCESSOR.DELETED, Boolean::valueOf);
        expressionMapper.map(ProcessorFields.UUID, PROCESSOR.UUID, value -> value);
    }

    @Override
    public Processor create(final Processor processor) {
        // We don't use the delegate DAO here as we want to handle potential duplicates carefully so this
        // behaves as a getOrCreate method.
        // TODO This should be replaced with JooqUtil.tryCreate as ANY error in the sql will be
        //  ignored, not just key duplicates.
        return JooqUtil.contextResult(processorDbConnProvider, context -> {
            final Optional<ProcessorRecord> optional = context
                    .insertInto(PROCESSOR,
                            PROCESSOR.CREATE_TIME_MS,
                            PROCESSOR.CREATE_USER,
                            PROCESSOR.UPDATE_TIME_MS,
                            PROCESSOR.UPDATE_USER,
                            PROCESSOR.UUID,
                            PROCESSOR.TASK_TYPE,
                            PROCESSOR.PIPELINE_UUID,
                            PROCESSOR.ENABLED,
                            PROCESSOR.DELETED)
                    .values(processor.getCreateTimeMs(),
                            processor.getCreateUser(),
                            processor.getUpdateTimeMs(),
                            processor.getUpdateUser(),
                            processor.getUuid(),
                            processor.getTaskType(),
                            processor.getPipelineUuid(),
                            processor.isEnabled(),
                            processor.isDeleted())
                    .onDuplicateKeyIgnore()
                    .returning(PROCESSOR.ID)
                    .fetchOptional();

            if (optional.isPresent()) {
                final Integer id = optional.get().getId();
                return context
                        .select()
                        .from(PROCESSOR)
                        .where(PROCESSOR.ID.eq(id))
                        .fetchOneInto(Processor.class);
            }

            return context
                    .select()
                    .from(PROCESSOR)
                    .where(PROCESSOR.PIPELINE_UUID.eq(processor.getPipelineUuid()))
                    .fetchOneInto(Processor.class);
        });
    }

    @Override
    public Processor update(final Processor processor) {
        return genericDao.update(processor);
    }

    @Override
    public boolean delete(final int id) {
        return genericDao.delete(id);
    }

    @Override
    public boolean logicalDelete(final int id) {
        try {
            return JooqUtil.transactionResult(processorDbConnProvider, context -> {

                // Logically delete any associated unprocessed tasks.
                // Once the filter is logically deleted no new tasks will be created for it
                // but we may still have active tasks for 'deleted' filters.
                final int processor_task_update_count = context
                        .update(PROCESSOR_TASK)
                        .set(PROCESSOR_TASK.STATUS, TaskStatus.DELETED.getPrimitiveValue())
                        .set(PROCESSOR_TASK.VERSION, PROCESSOR_TASK.VERSION.plus(1))
                        .where(DSL.exists(
                                DSL.selectZero()
                                        .from(PROCESSOR_FILTER)
                                        .innerJoin(PROCESSOR)
                                        .on(PROCESSOR_FILTER.FK_PROCESSOR_ID.eq(PROCESSOR.ID))
                                        .where(PROCESSOR.ID.eq(id))))
                        .and(PROCESSOR_TASK.STATUS.eq(TaskStatus.UNPROCESSED.getPrimitiveValue()))
                        .execute();

                LOGGER.debug("Logically deleted {} tasks for processor Id {}",
                        processor_task_update_count, id);

                // Logically delete all the child filters first
                final int processor_filter_update_count = context
                        .update(PROCESSOR_FILTER)
                        .set(PROCESSOR_FILTER.DELETED, true)
                        .set(PROCESSOR_FILTER.VERSION, PROCESSOR_FILTER.VERSION.plus(1))
                        .where(PROCESSOR_FILTER.FK_PROCESSOR_ID.eq(id))
                        .execute();
                LOGGER.debug("Logically deleted {} filters for processor Id {}",
                        processor_filter_update_count, id);

                final int processor_update_count = context
                        .update(PROCESSOR)
                        .set(PROCESSOR.DELETED, true)
                        .set(PROCESSOR.VERSION, PROCESSOR.VERSION.plus(1))
                        .where(PROCESSOR.ID.eq(id))
                        .execute();

                LOGGER.debug("Logically deleted {} processor with Id {}",
                        processor_update_count, id);

                return processor_update_count > 0;
            });
        } catch (Exception e) {
            throw new RuntimeException(
                    "Error deleting filters, tasks and processor for processor id " + id, e);
        }
    }

    @Override
{'fraction_non_alphanumeric': 0.07879213483146068, 'fraction_numerical': 0.0009129213483146068, 'mean_word_length': 1.7950932286555445, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3322023', 'n_tokens_mistral': 3544, 'n_tokens_neox': 3484, 'n_words': 675}