Service.kt

/*
 * fedora-ingest-service
 * Copyright (C) 2020 Memoriav
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

package org.memobase

import ch.memobase.exceptions.SftpClientException
import ch.memobase.fedora.FedoraClient
import ch.memobase.fedora.FedoraClientImpl
import ch.memobase.reporting.ReportStatus
import ch.memobase.settings.SettingsLoader
import ch.memobase.sftp.SftpClient
import java.io.Closeable
import java.io.IOException
import java.net.URISyntaxException
import java.util.Properties
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.logging.log4j.LogManager
import org.apache.logging.log4j.Logger
import org.fcrepo.client.FcrepoOperationFailedException
import org.memobase.exceptions.MissingMimeTypeException

class Service(fileName: String = "app.yml") : Closeable {

    companion object {
        const val FEDORA_PROPERTIES_PREFIX = "fedora"
        const val CONSUMER_MAX_POLL_RECORDS = "10"
        const val CONSUMER_MAX_INTERVAL_MS = "600000" // 600000ms = 10min
        const val BINARY_FILE_URI_PATH = "binary"
        const val SFTP_PREFIX = "sftp:"

        fun createFedoraClient(appSettings: Properties): FedoraClient {
            return FedoraClientImpl.builder()
                .properties(appSettings, FEDORA_PROPERTIES_PREFIX)
                .build()
        }
    }

    private val settings = SettingsLoader(
        listOf(
            "isSimple",
            "$FEDORA_PROPERTIES_PREFIX.internalBaseUrl",
            "$FEDORA_PROPERTIES_PREFIX.externalBaseUrl",
            "$FEDORA_PROPERTIES_PREFIX.username",
            "$FEDORA_PROPERTIES_PREFIX.password"
        ),
        fileName,
        useProducerConfig = true,
        useConsumerConfig = true,
        readSftpSettings = true
    )

    private val log: Logger = LogManager.getLogger("FedoraIngestService")
    private val fedoraClient = createFedoraClient(settings.appSettings)
    private val simpleIngester = SimpleIngester(fedoraClient)
    private val isSimple = (settings.appSettings.getProperty("isSimple") ?: "false").toBoolean()
    private var consumer: Consumer
    private var producer: Producer
    private var sftpClient: SftpClient? = null

    init {
        val consumerSettings = settings.kafkaConsumerSettings
        consumerSettings.setProperty("max.poll.records", CONSUMER_MAX_POLL_RECORDS)
        consumerSettings.setProperty("max.poll.interval.ms", CONSUMER_MAX_INTERVAL_MS)
        consumer = Consumer(consumerSettings, settings.inputTopic)
        producer = Producer(settings.kafkaProducerSettings, settings.processReportTopic)
        log.info("Connected to Kafka cluster.")
        if (!isSimple) {
            sftpClient = SftpClient(settings.sftpSettings)
            log.info("Connected to sFTP server.")
        }
    }

    fun run() {
        while (true) {
            processRecords()
        }
    }

    fun processRecords() {
        for (record in consumer.fetchRecords()) {
            val ingestReport = if (isSimple)
                processSingleEntity(record)
            else
                processRecord(record)
            producer.sendReport(record.headers(), ingestReport)
        }
    }

    /**
     * If there is only a single named entity within the content data and no binaries are loaded.
     */
    private fun processSingleEntity(record: ConsumerRecord<String, String>): Report {
        val key = record.key()
        val value = record.value()

        if (key == null)
            return Report("NoKey", ReportStatus.fatal, "The key in message is null.")

        if (value == null)
            return Report(key, ReportStatus.fatal, "The value in message is null.")

        return try {
            simpleIngester.ingest(key, value)
            Report(
                key, ReportStatus.success, "Successfully ingested message into Fedora repository."
            )
        } catch (ex: FcrepoOperationFailedException) {
            log.error(ex.localizedMessage)
            Report(
                id = record.key(),
                status = ReportStatus.fatal,
                message = "Unable to write to Fedora: ${ex.localizedMessage}"
            )
        } catch (ex: IOException) {
        log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = record.key(),
                status = ReportStatus.fatal,
                message = "IO error: ${ex.localizedMessage}"
            )
        } catch (ex: URISyntaxException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = record.key(),
                status = ReportStatus.fatal,
                message = "Malformed URI: ${ex.localizedMessage}"
            )
        } catch (ex: Exception) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = record.key(),
                status = ReportStatus.fatal,
                message = "Unknown error: ${ex.localizedMessage}"
            )
        }
    }

    private fun processRecord(record: ConsumerRecord<String, String>): Report {
        val key = record.key()
        val value = record.value()

        if (key == null)
            return Report("NoKey", ReportStatus.fatal, "The key in message is null.")

        if (value == null)
            return Report(key, ReportStatus.fatal, "The value in message is null.")

        return try {
            val ingester = sftpClient?.let {
                Ingester(
                    it,
                    fedoraClient,
                    settings.appSettings.getProperty("$FEDORA_PROPERTIES_PREFIX.externalBaseUrl")
                )
            }
            ingester?.ingest(key, value)
            Report(
                id = key,
                status = ReportStatus.success,
                message = ReportMessages.ingestedRecord(key)
            )
        } catch (ex: FcrepoOperationFailedException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "Unable to write to Fedora: ${ex.localizedMessage}"
            )
        } catch (ex: IOException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "IO error: ${ex.localizedMessage}"
            )
        } catch (ex: MissingMimeTypeException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "Missing MimeType: ${ex.localizedMessage}"
            )
        } catch (ex: SftpClientException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "Unable to read from SFTP: ${ex.localizedMessage}"
            )
        } catch (ex: URISyntaxException) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "Malformed URI: ${ex.localizedMessage}"
            )
        } catch (ex: Exception) {
            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
            Report(
                id = key,
                status = ReportStatus.fatal,
                message = "Unknown error: ${ex.localizedMessage}"
            )
        }
    }

    override fun close() {
        consumer.close()
        producer.close()
        sftpClient?.close()
    }
}