main.nf

#!/usr/bin/env nextflow
/*
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    nds-lucid/graphdbloader
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    Github : https://github.com/nds-lucid/graphdbloader
----------------------------------------------------------------------------------------
*/
nextflow.enable.dsl = 2


include { graphdb_import_file; graphdb_auth } from './modules/graphdb'
include { clean_backups } from './modules/io'
include { copy_file as copy_to_graphdb } from './modules/io'

/*
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    RUN ALL WORKFLOWS
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*/

//
// WORKFLOW: Execute a single named workflow for the pipeline
// See: https://github.com/nf-core/rnaseq/issues/619
//
workflow {
    // Copy input files from source to GraphDB
    if (params.watch)
        ch_input_files = Channel.watchPath("${params.input_dir}/*.nt.gz")
    else
        ch_input_files = Channel.fromPath("${params.input_dir}/*.nt.gz")
    
    ch_server_files = copy_to_graphdb(ch_input_files, params.graphdb_dir)

    // Get GraphDB auth token using nextflow secrets
    token = graphdb_auth()

    // Extract graph uri from filename, assuming format {timstamp}_{graph}.{ext}
    // NOTE: uses baseName+replace instead of simpleName to prevent truncation of
    // iso8601-compliant timestamps with milliseconds
    ch_graph_uri = ch_input_files
      .map {
        it
          .baseName
          .tokenize('_')
          .last()
          .replaceAll(/\..*$/, '')
      }

    // Load into GraphDB using server file import and copy to backup dir.
    ch_graphdb = graphdb_import_file(
      ch_server_files,
      ch_graph_uri,
      token
    )

    // Log successful and failed imports (HTTP codes in 2xx range).
    ch_log = ch_graphdb.map {
        file, graph, resp, code -> [
          file: file,
          graph: graph,
          resp: resp.text
            .replaceAll('[\n\r]', '')
            .replaceAll('"', '\\\\"'),
          status: code.text ==~ /2[0-9]{2}/ ? "SUCCESS" : "FAILURE"
        ]
      }

    // Send event to log_dir/timestamp.json
    def log_event = { event ->
        new File( params.log_dir + "/${new Date().toInstant().now()}.json" ) << event + "\n"
    }

    // Write all import events to disk as json (successes and failures).
    ch_log
      .map {
          it -> "{\"status\": \"${it.status}\", \"file\": \"${it.file}\", \"graph\": \"${it.graph}\", \"resp\": \"${it.resp}\"}"
        }
      .subscribe { log_event(it) }

    // List all successfully imported files.
    ch_imported_files = ch_log
      .filter { it.status == "SUCCESS" }
      .map { it.file }

    // Only retain N last backups for each graph, if set
    if ( params.max_backups >= 1 ) {
      // Only retain N last backups for each graph, if set
      // passing channel to trigger process per file in watchPath
      clean_backups(params.backup_dir, params.max_backups, ch_imported_files)
    }


}


workflow.onError {
    println "Error: ${workflow.errorMessage}"
    println "Error: ${workflow.errorReport}"
}

/*
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    THE END
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*/