class DatabaseFDFlowManager extends BaseService implements FDFlowManager {

    @Override
    public FDFlowEvent addFlowEvent(final FDFlowEvent flowEvent, final VersionedProcessGroup flowContent, final NiFiUser user) {
        if (flowEvent == null) {
            throw new IllegalArgumentException("Flow event cannot be null or blank");
        }

        if (StringUtils.isBlank(flowEvent.getFlowIdentifier())) {
            throw new IllegalArgumentException("Flow identifier of flow event cannot be null or blank");
        }

        if (flowContent == null) {
            throw new IllegalArgumentException("Flow content cannot be null or blank");
        }

        if (StringUtils.isBlank(flowContent.getIdentifier()) || StringUtils.isBlank(flowContent.getName())) {
            throw new IllegalArgumentException("Flow content must contain a root process group with an identifier and name");
        }

        if (user == null || StringUtils.isBlank(user.getIdentity())) {
            throw new IllegalArgumentException("A user identity must be provided");
        }

        final Optional<FDFlowEntity> flowEntity = flowRepository.findById(flowEvent.getFlowIdentifier());
        if (!flowEntity.isPresent()) {
            throw new ResourceNotFoundException("No flow exists with the provided flow id");
        }

        final String previousRootGroupId = flowEntity.get().getRootProcessGroupId();
        final String incomingRootGroupId = flowContent.getIdentifier();
        if (!previousRootGroupId.equals(incomingRootGroupId)) {
            throw new IllegalArgumentException("Cannot change the id of the root process group");
        }

        final Optional<FDFlowEventEntity> latestFlowEvent = flowEventRepository.findFirstByFlowOrderByFlowRevisionDesc(flowEntity.get());
        if (!latestFlowEvent.isPresent()) {
            throw new IllegalStateException("Could not find any flow content for the given flow");
        }

        final BigInteger currRevision = latestFlowEvent.get().getFlowRevision();
        final BigInteger nextRevision = currRevision.add(BigInteger.ONE);

        flowEvent.setIdentifier(UUID.randomUUID().toString());
        flowEvent.setCreated(System.currentTimeMillis());
        flowEvent.setUpdated(System.currentTimeMillis());
        flowEvent.setFlowRevision(nextRevision);
        flowEvent.setFlowFormat(CURRENT_FLOW_FORMAT);
        flowEvent.setUserIdentity(user.getIdentity());

        // if not a publish event then carry forward the registry info from the previous event, otherwise use whatever was specified
        if (flowEvent.getEventType() != FDFlowEventType.FLOW_PUBLISHED) {
            flowEvent.setRegistryUrl(latestFlowEvent.get().getRegistryUrl());
            flowEvent.setRegistryBucketId(latestFlowEvent.get().getRegistryBucketId());
            flowEvent.setRegistryFlowId(latestFlowEvent.get().getRegistryFlowId());
            flowEvent.setRegistryVersion(latestFlowEvent.get().getRegistryVersion());

            final Date lastPublished = latestFlowEvent.get().getLastPublished();
            flowEvent.setLastPublished(lastPublished == null ? null : lastPublished.getTime());
            flowEvent.setLastPublishedUserIdentity(latestFlowEvent.get().getLastPublishedUserIdentity());
        }

        if (flowEvent.getEventType() == FDFlowEventType.FLOW_PUBLISHED && flowEvent.getLastPublished() == null) {
            throw new IllegalArgumentException("Publish event must contain a published date");
        }

        validate(flowEvent, "Cannot add flow event due to invalid event");

        final FDFlowEventEntity flowEventEntity = modelMapper.map(flowEvent, FDFlowEventEntity.class);
        flowEventEntity.setFlowContent(serializeFlowContent(flowContent));
        flowEventEntity.setFlow(flowEntity.get());

        final FDFlowEventEntity savedFlowEventEntity = flowEventRepository.save(flowEventEntity);

        // Update the cached flow with the latest version info
        final String flowId = flowEntity.get().getId();
        if (flowsById.containsKey(flowId)) {
            final FDVersionInfo versionInfo = createVersionInfo(savedFlowEventEntity);
            final FDFlow cachedFlow = flowsById.get(flowId);
            cachedFlow.setVersionInfo(versionInfo);
            cachedFlow.setLocalFlowRevision(savedFlowEventEntity.getFlowRevision());
        }

        return modelMapper.map(savedFlowEventEntity, FDFlowEvent.class);
    }

    @Autowired public  DatabaseFDFlowManager(final FDFlowRepository flowRepository,
                                 final FDFlowEventRepository flowEventRepository,
                                 final OptionalModelMapper modelMapper,
                                 final ObjectMapper objectMapper,
                                 final Validator validator);

}