Ingester.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 java.io.File
import java.io.StringWriter
import java.net.URI
import org.apache.jena.rdf.model.Model
import org.apache.jena.riot.Lang
import org.apache.jena.riot.RDFDataMgr
import org.apache.logging.log4j.LogManager
import org.fcrepo.client.FcrepoOperationFailedException
import org.memobase.fedora.FedoraClient
import org.memobase.fedora.FedoraTransactionClient
import org.memobase.fedora.RdfContentTypes
import org.memobase.sftp.SftpClient

class Ingester(
    private val sftpClient: SftpClient,
    private val fedoraClient: FedoraClient,
    private val externalBaseUrl: String
) {

    private val log = LogManager.getLogger("FedoraIngester")

    @Throws(FcrepoOperationFailedException::class)
    fun ingest(id: String, content: String) {
        log.info("Begin ingest of message with id $id.")
        val rdfHandler = RdfHandler(content, externalBaseUrl)
        val recordOutput = StringWriter()
        val recordPair = rdfHandler.getRecord()
        RDFDataMgr.write(recordOutput, recordPair.second, Lang.NTRIPLES)
        val data = recordOutput.toString()

        fedoraClient.startTransaction().use { transaction ->
            log.info("Ingesting record ${recordPair.first}.")

            // create placeholders referenced resources:
            val nonBinaryResources = rdfHandler.getReferencedNonBinaryResources()
            nonBinaryResources.forEach { resource ->
                log.info("Creating placeholder for resource $resource.")
                transaction.createPlaceholder(URI(resource))
                log.info("Created placeholder for resource $resource.")
            }

            // ingest record, instantiations and binaries:
            transaction.createOrUpdateRdfResource(URI(recordPair.first), data, RdfContentTypes.NTRIPLES)
            ingestInstantiations(rdfHandler.getInstantiations(), transaction)
            val sftpLocators = rdfHandler.getSftpLocators()
            if (sftpLocators.isNotEmpty()) {
                ingestBinaries(sftpLocators, rdfHandler, transaction)
            }
            transaction.commit()
            log.info("Ingested record ${recordPair.first}.")
        }
    }

    private fun ingestInstantiations(instantiations: List<Pair<String, Model>>, transaction: FedoraTransactionClient) {
        instantiations.forEach { instantiationPair ->
            val instantiationOutput = StringWriter()
            RDFDataMgr.write(instantiationOutput, instantiationPair.second, Lang.NTRIPLES)
            val instantiationData = instantiationOutput.toString()
            log.info("Ingesting instantiation ${instantiationPair.first}.")
            transaction.createOrUpdateRdfResource(URI(instantiationPair.first), instantiationData, RdfContentTypes.NTRIPLES)
            log.info("Ingested instantiation ${instantiationPair.first}.")
        }
    }

    private fun ingestBinaries(sftpLocators: List<Pair<String, String?>>, rdfHandler: RdfHandler, transaction: FedoraTransactionClient) {
        sftpLocators.forEach {
            val digitalInstantiationUrl = it.first
            it.second.let { path ->
                if (path != null) {
                    sftpClient.open(File(path)).use { stream ->
                        val binaryUri = "$digitalInstantiationUrl/${Service.BINARY_FILE_URI_PATH}"
                        val mimeType = rdfHandler.getMimeType(digitalInstantiationUrl)
                        log.info("Ingesting binary $binaryUri with mime type $mimeType.")
                        transaction.createOrUpdateBinaryResource(URI(binaryUri), stream, mimeType)
                    }
                }
            }
        }
    }
}