class DatabaseFDFlowManager extends BaseService implements FDFlowManager {

    @Override
    public Optional<FDFlow> getFlow(final String flowId) {
        if (StringUtils.isBlank(flowId)) {
            throw new IllegalArgumentException("Flow id cannot be null or blank");
        }

        // attempt to retrieve from cache
        FDFlow flow = flowsById.get(flowId);

        // if not cached then load fresh and put in cache
        if (flow == null) {
            final Optional<FDFlowEntity> flowEntity = flowRepository.findById(flowId);
            if (!flowEntity.isPresent()) {
                return Optional.empty();
            }

            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 FDFlowMetadata flowMetadata = modelMapper.map(flowEntity.get(), FDFlowMetadata.class);
            final VersionedProcessGroup flowContent = deserialize(latestFlowEvent.get().getFlowContent());
            final FDVersionInfo versionInfo = createVersionInfo(latestFlowEvent.get());
            final BigInteger localFlowRevision = latestFlowEvent.get().getFlowRevision();

            flow = new FDFlow();
            flow.setFlowMetadata(flowMetadata);
            flow.setFlowContent(flowContent);
            flow.setVersionInfo(versionInfo);
            flow.setLocalFlowRevision(localFlowRevision);

            flowsById.put(flowId, flow);
        }

        return Optional.of(flow);
    }

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

}