/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */
package org.apache.causeway.extensions.commandlog.applib.dom;

import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;

import jakarta.inject.Inject;
import jakarta.inject.Provider;

import org.springframework.lang.Nullable;

import org.apache.causeway.applib.jaxb.JavaSqlXMLGregorianCalendarMarshalling;
import org.apache.causeway.applib.query.Query;
import org.apache.causeway.applib.query.QueryRange;
import org.apache.causeway.applib.services.bookmark.Bookmark;
import org.apache.causeway.applib.services.command.Command;
import org.apache.causeway.applib.services.factory.FactoryService;
import org.apache.causeway.applib.services.repository.RepositoryService;
import org.apache.causeway.applib.util.schema.CommandDtoUtils;
import org.apache.causeway.commons.internal.base._Casts;
import org.apache.causeway.core.config.environment.CausewaySystemEnvironment;
import org.apache.causeway.schema.cmd.v2.CommandDto;
import org.apache.causeway.schema.cmd.v2.CommandsDto;
import org.apache.causeway.schema.cmd.v2.MapDto;
import org.apache.causeway.schema.common.v2.InteractionType;

import lombok.val;

/**
 * Provides supporting functionality for querying {@link CommandLogEntry command log entry} entities.
 *
 * @since 2.0 {@index}
 */
public abstract class CommandLogEntryRepositoryAbstract<C extends CommandLogEntry> implements CommandLogEntryRepository {

    @Inject Provider<RepositoryService> repositoryServiceProvider;
    @Inject FactoryService factoryService;
    @Inject CausewaySystemEnvironment causewaySystemEnvironment;

    private final Class<C> commandLogEntryClass;


    protected CommandLogEntryRepositoryAbstract(final Class<C> commandLogEntryClass) {
        this.commandLogEntryClass = commandLogEntryClass;
    }

    public Class<C> getEntityClass() {
        return commandLogEntryClass;
    }

    public C createEntryAndPersist(
            final Command command, final UUID parentInteractionIdIfAny, final ExecuteIn executeIn) {
        C c = factoryService.detachedEntity(commandLogEntryClass);
        c.sync(command);
        c.setParentInteractionId(parentInteractionIdIfAny);
        c.setExecuteIn(executeIn);
        persist(c);
        return c;
    }

    @Override
    public Optional<CommandLogEntry> findByInteractionId(final UUID interactionId) {
        return _Casts.uncheckedCast(
                repositoryService().firstMatch(
                Query.named(commandLogEntryClass,  CommandLogEntry.Nq.FIND_BY_INTERACTION_ID)
                        .withParameter("interactionId", interactionId))
        );
    }

    @Override
    public List<CommandLogEntry> findByParent(final CommandLogEntry parent) {
        return findByParentInteractionId(parent.getInteractionId());
    }

    @Override
    public List<CommandLogEntry> findByParentInteractionId(final UUID parentInteractionId) {
        return _Casts.uncheckedCast( repositoryService().allMatches(
                Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_PARENT_INTERACTION_ID)
                        .withParameter("parentInteractionId", parentInteractionId))
        );
    }

    @Override
    public List<CommandLogEntry> findByFromAndTo(
            final @Nullable LocalDate from,
            final @Nullable LocalDate to) {
        final Timestamp fromTs = toTimestampStartOfDayWithOffset(from, 0);
        final Timestamp toTs = toTimestampStartOfDayWithOffset(to, 1);

        final Query<C> query;
        if(from != null) {
            if(to != null) {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TIMESTAMP_BETWEEN)
                        .withParameter("from", fromTs)
                        .withParameter("to", toTs);
            } else {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TIMESTAMP_AFTER)
                        .withParameter("from", fromTs);
            }
        } else {
            if(to != null) {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TIMESTAMP_BEFORE)
                        .withParameter("to", toTs);
            } else {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND);
            }
        }
        return _Casts.uncheckedCast(repositoryService().allMatches(query));
    }

    @Override
    public List<CommandLogEntry> findCurrent() {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_CURRENT))
        );
    }

    @Override
    public List<CommandLogEntry> findCompleted() {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_COMPLETED))
        );
    }


    public List<CommandLogEntry> findByTargetAndFromAndTo(
            final Bookmark target,
            final @Nullable LocalDate from,
            final @Nullable LocalDate to) {

        val fromTs = toTimestampStartOfDayWithOffset(from, 0);
        val toTs = toTimestampStartOfDayWithOffset(to, 1);

        final Query<C> query;
        if(from != null) {
            if(to != null) {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TARGET_AND_TIMESTAMP_BETWEEN)
                        .withParameter("target", target)
                        .withParameter("from", fromTs)
                        .withParameter("to", toTs);
            } else {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TARGET_AND_TIMESTAMP_AFTER)
                        .withParameter("target", target)
                        .withParameter("from", fromTs);
            }
        } else {
            if(to != null) {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TARGET_AND_TIMESTAMP_BEFORE)
                        .withParameter("target", target)
                        .withParameter("to", toTs);
            } else {
                query = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_TARGET)
                        .withParameter("target", target);
            }
        }
        return _Casts.uncheckedCast(repositoryService().allMatches(query));
    }


    @Override
    public List<CommandLogEntry> findMostRecent() {
        return findMostRecent(100);
    }

    @Override
    public List<CommandLogEntry> findMostRecent(final int limit) {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                Query.named(commandLogEntryClass,  CommandLogEntry.Nq.FIND_MOST_RECENT).withLimit(limit))
        );
    }


    @Override
    public List<CommandLogEntry> findRecentByUsername(final String username) {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_RECENT_BY_USERNAME)
                        .withParameter("username", username)
                        .withLimit(30L))
        );
    }



    public List<CommandLogEntry> findRecentByTarget(final Bookmark target) {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_RECENT_BY_TARGET)
                        .withParameter("target", target)
                        .withLimit(30L)
                )
        );
    }

    public List<CommandLogEntry> findRecentByTargetOrResult(final Bookmark targetOrResult) {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_RECENT_BY_TARGET_OR_RESULT)
                            .withParameter("targetOrResult", targetOrResult)
                            .withLimit(30L)
                    )
        );
    }


    /**
     * Intended to support the replay of commands on a secondary instance of
     * the application.
     *
     * This finder returns all (completed) {@link CommandLogEntry}s started after
     * the command with the specified interactionId.  The number of commands
     * returned can be limited so that they can be applied in batches.
     *
     * If the provided interactionId is null, then only a single
     * {@link CommandLogEntry command} is returned.  This is intended to support
     * the case when the secondary does not yet have any
     * {@link CommandLogEntry command}s replicated.  In practice this is unlikely;
     * typically we expect that the secondary will be set up to run against a
     * copy of the primary instance's DB (restored from a backup), in which
     * case there will already be a {@link CommandLogEntry command} representing the
     * current high water mark on the secondary system.
     *
     * If the interactionId is not null but the corresponding
     * {@link CommandLogEntry command} is not found, then <tt>null</tt> is returned.
     * In the replay scenario the caller will probably interpret this as an
     * error because it means that the high water mark on the secondary is
     * inaccurate, referring to a non-existent {@link CommandLogEntry command} on
     * the primary.
     *
     * @param interactionId - the identifier of the {@link CommandLogEntry command} being
     *                   the replay HWM (using {@link #findMostRecentReplayed()} on the
     *                   secondary), or null if no HWM was found there.
     * @param batchSize - to restrict the number returned (so that replay
     *                   commands can be batched).
     */
    @Override
    public List<CommandLogEntry> findSince(final UUID interactionId, final Integer batchSize) {
        if(interactionId == null) {
            return findFirst();
        }
        final C from = findByInteractionIdElseNull(interactionId);
        if(from == null) {
            return Collections.emptyList();
        }
        return findSince(from.getTimestamp(), batchSize);
    }

    private List<CommandLogEntry> findFirst() {
        Optional<CommandLogEntry> firstCommandIfAny =
                _Casts.uncheckedCast(repositoryService().firstMatch(
                Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_FIRST)));
        return firstCommandIfAny
                .map(Collections::singletonList)
                .orElse(Collections.emptyList());
    }

    /**
     * Returns any persisted commands that have not yet started.
     *
     * <p>
     * This is to support the notion of background commands (the same as their implementation in v1) whereby a
     * custom executor service for {@link org.apache.causeway.applib.services.wrapper.WrapperFactory} would
     * &quot;execute&quot; a {@link Command} simply by persisting it as a {@link CommandLogEntry}, so that a
     * quartz or similar background job could execute the {@link Command} at some point later.
     * </p>
     */
    @Override
    public List<CommandLogEntry> findBackgroundAndNotYetStarted() {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BACKGROUND_AND_NOT_YET_STARTED)));
    }

    public List<CommandLogEntry> findRecentBackgroundByTarget(final Bookmark target) {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_RECENT_BACKGROUND_BY_TARGET)
                        .withParameter("target", target)
                        .withLimit(30L)
                )
        );
    }


    /**
     * The most recent replayed command previously replicated from primary to
     * secondary.
     *
     * <p>
     * This should always exist except for the very first times
     * (after restored the prod DB to secondary).
     * </p>
     */
    @Override
    public Optional<CommandLogEntry> findMostRecentReplayed() {
        return _Casts.uncheckedCast(
                repositoryService().firstMatch(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_MOST_RECENT_REPLAYED))
        );
    }

    /**
     * The most recent completed command, as queried on the
     * secondary.
     *
     * <p>
     *     After a restart following the production database being restored
     *     from primary to secondary, would correspond to the last command
     *     run on primary before the production database was restored to the
     *     secondary.
     * </p>
     */
    @Override
    public Optional<CommandLogEntry> findMostRecentCompleted() {
        return _Casts.uncheckedCast(
                repositoryService().firstMatch(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_MOST_RECENT_COMPLETED))
        );
    }

    @Override
    public List<CommandLogEntry> findNotYetReplayed() {
        return _Casts.uncheckedCast(
                repositoryService().allMatches(
                    Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_REPLAY_STATE)
                        .withParameter("replayState", ReplayState.PENDING)
                        .withLimit(10))
        );
    }


    public C saveForReplay(final CommandDto dto) {

        if(dto.getMember().getInteractionType() == InteractionType.ACTION_INVOCATION) {
            final MapDto userData = dto.getUserData();
            if (userData == null ) {
                throw new IllegalStateException(String.format(
                        "Can only persist action DTOs with additional userData; got: \n%s",
                        CommandDtoUtils.dtoMapper().toString(dto)));
            }
        }

        final C commandJdo = factoryService.detachedEntity(commandLogEntryClass);

        commandJdo.setInteractionId(UUID.fromString(dto.getInteractionId()));
        commandJdo.setTimestamp(JavaSqlXMLGregorianCalendarMarshalling.toTimestamp(dto.getTimestamp()));
        commandJdo.setUsername(dto.getUsername());

        commandJdo.setReplayState(ReplayState.PENDING);

        val firstTargetOidDto = dto.getTargets().getOid().get(0);
        commandJdo.setTarget(Bookmark.forOidDto(firstTargetOidDto));
        commandJdo.setCommandDto(dto);
        commandJdo.setLogicalMemberIdentifier(dto.getMember().getLogicalMemberIdentifier());

        persist(commandJdo);

        return commandJdo;
    }


    public List<CommandLogEntry> saveForReplay(final CommandsDto commandsDto) {
        val commandDtos = commandsDto.getCommandDto();
        val commands = new ArrayList<CommandLogEntry>();
        for (val dto : commandDtos) {
            commands.add(saveForReplay(dto));
        }
        return commands;
    }


    @Override
    public void persist(final CommandLogEntry commandLogEntry) {
        repositoryService().persistAndFlush(commandLogEntry);
    }

    @Override
    public void truncateLog() {
        repositoryService().removeAll(commandLogEntryClass);
    }

    // --


    @Override
    public List<CommandLogEntry> findCommandsOnPrimaryElseFail(
            final @Nullable UUID interactionId,
            final @Nullable Integer batchSize) throws NotFoundException {

        final List<CommandLogEntry> commands = findSince(interactionId, batchSize);
        if(commands == null) {
            throw new NotFoundException(interactionId);
        }
        return commands;
    }



    private C findByInteractionIdElseNull(final UUID interactionId) {
        val q = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_BY_INTERACTION_ID)
                .withParameter("interactionId", interactionId);
        return repositoryService().uniqueMatch(q).orElse(null);
    }

    private List<CommandLogEntry> findSince(
            final Timestamp timestamp,
            final Integer batchSize) {

        // DN generates incorrect SQL for SQL Server if count set to 1; so we set to 2 and then trim
        // XXX that's a historic workaround, should rather be fixed upstream
        val needsTrimFix = batchSize != null && batchSize == 1;

        val q = Query.named(commandLogEntryClass, CommandLogEntry.Nq.FIND_SINCE)
                .withParameter("timestamp", timestamp)
                .withRange(QueryRange.limit(
                        needsTrimFix ? 2L : batchSize
                ));

        final List<CommandLogEntry> commandJdos = _Casts.uncheckedCast(repositoryService().allMatches(q));
        return needsTrimFix && commandJdos.size() > 1
                ? commandJdos.subList(0,1)
                : commandJdos;
    }





    private RepositoryService repositoryService() {
        return repositoryServiceProvider.get();
    }

    private static Timestamp toTimestampStartOfDayWithOffset(
            final @Nullable LocalDate dt,
            final int daysOffset) {

        return dt!=null
                ? new java.sql.Timestamp(
                Instant.from(dt.atStartOfDay().plusDays(daysOffset).atZone(ZoneId.systemDefault()))
                        .toEpochMilli())
                : null;
    }

    /**
     * intended for testing purposes only
     */
    @Override
    public List<CommandLogEntry> findAll() {
        if (causewaySystemEnvironment.getDeploymentType().isProduction()) {
            throw new IllegalStateException("Cannot call 'findAll' in production systems");
        }
        return _Casts.uncheckedCast(repositoryService().allInstances(commandLogEntryClass));
    }


    /**
     * intended for testing purposes only
     */
    @Override
    public void removeAll() {
        if (causewaySystemEnvironment.getDeploymentType().isProduction()) {
            throw new IllegalStateException("Cannot call 'removeAll' in production systems");
        }
        repositoryService().removeAll(commandLogEntryClass);
    }


}
