refactor(nf-tower): Replace manual HTTP with tower-java-sdk for dataset upload

Refactor dataset upload implementation to use the official tower-java-sdk
instead of manual HTTP multipart encoding, significantly simplifying the
code and improving maintainability.

Changes:
- Add tower-java-sdk dependency (1.43.1) with GitHub Packages repository
- Replace manual HTTP implementation with DatasetsApi SDK methods:
  - createDataset() now uses datasetsApi.createDataset(wspId, request)
  - uploadIndexToDataset() now uses datasetsApi.uploadDataset(wspId, id, header, file)
- Remove ~120 lines of manual HTTP code:
  - Deleted getUrlDatasets() and getUrlDatasetUpload() URL builders
  - Deleted uploadFile() multipart HTTP request construction
  - Deleted createMultipartBody() RFC 2388 multipart encoding
- Add comprehensive test coverage:
  - 7 unit tests with mocked DatasetsApi (initialization, event collection, 
    dataset creation, file upload, exception handling)
  - 1 integration test with real Platform API (conditional on TOWER_ACCESS_TOKEN)
  - Manual test workflow in test-dataset-upload/ directory with documentation

Testing:
- All unit tests passing (BUILD SUCCESSFUL)
- Integration test ready (runs when TOWER_ACCESS_TOKEN available)
- Test workflow provides end-to-end validation guide

Benefits:
- Uses official Seqera SDK (same as tower-cli)
- Easier to test with mocked API
- SDK handles all HTTP/multipart details automatically
- Bug fixes in SDK benefit us automatically
- Code reduced from ~300 lines to ~100 lines

Note: Requires GitHub credentials for tower-java-sdk dependency.
Configure github_username and github_access_token in gradle.properties
or set GITHUB_USERNAME and GITHUB_TOKEN environment variables.

Signed-off-by: Edmund Miller <[email protected]>
Signed-off-by: Edmund Miller <[email protected]>

Author: edmundmiller

plugins/nf-tower/build.gradle

@@ -30,6 +30,16 @@ nextflowPlugin {
     ]
 }
 
+repositories {
+    maven {
+        url = uri("https://maven.pkg.github.com/seqeralabs/tower-java-sdk")
+        credentials {
+            username = project.findProperty('github_username') ?: System.getenv("GITHUB_USERNAME")
+            password = project.findProperty('github_access_token') ?: System.getenv("GITHUB_TOKEN")
+        }
+    }
+}
+
 sourceSets {
     main.java.srcDirs = []
     main.groovy.srcDirs = ['src/main']
@@ -50,6 +60,7 @@ dependencies {
     compileOnly 'org.pf4j:pf4j:3.12.0'
     compileOnly 'io.seqera:lib-httpx:2.1.0'
 
+    api 'io.seqera.tower:tower-java-sdk:1.43.1'
     api "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.15.0"
     api "com.fasterxml.jackson.core:jackson-databind:2.12.7.1"
 

plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy

@@ -35,6 +35,11 @@ import groovy.transform.TupleConstructor
 import groovy.util.logging.Slf4j
 import io.seqera.http.HxClient
 import io.seqera.util.trace.TraceUtils
+import io.seqera.tower.ApiClient
+import io.seqera.tower.ApiException
+import io.seqera.tower.api.DatasetsApi
+import io.seqera.tower.model.CreateDatasetRequest
+import io.seqera.tower.model.CreateDatasetResponse
 import nextflow.BuildInfo
 import nextflow.Session
 import nextflow.container.resolver.ContainerMeta
@@ -101,6 +106,8 @@ class TowerClient implements TraceObserverV2 {
 
     private HxClient httpClient
 
+    private DatasetsApi datasetsApi
+
     private JsonGenerator generator
 
     private String workflowId
@@ -157,6 +164,7 @@ class TowerClient implements TraceObserverV2 {
         this.schema = loadSchema()
         this.generator = TowerJsonGenerator.create(schema)
         this.reports = new TowerReports(session)
+        this.datasetsApi = createDatasetsApi()
     }
 
     TowerClient withEnvironment(Map env) {
@@ -169,6 +177,30 @@ class TowerClient implements TraceObserverV2 {
         this.generator = TowerJsonGenerator.create(Collections.EMPTY_MAP)
     }
 
+    /**
+     * Create and configure a DatasetsApi client for Seqera Platform
+     *
+     * @return Configured DatasetsApi instance
+     */
+    protected DatasetsApi createDatasetsApi() {
+        if( !accessToken || !endpoint ) {
+            return null
+        }
+
+        try {
+            def apiClient = new ApiClient()
+            apiClient.setBasePath(endpoint)
+            apiClient.setBearerToken(accessToken)
+            apiClient.setUserAgent("Nextflow/$BuildInfo.version")
+
+            return new DatasetsApi(apiClient)
+        }
+        catch( Exception e ) {
+            log.warn "Failed to initialize DatasetsApi: ${e.message}"
+            return null
+        }
+    }
+
     @Override
     boolean enableMetrics() { true }
 
@@ -266,18 +298,6 @@ class TowerClient implements TraceObserverV2 {
         return result
     }
 
-    protected String getUrlDatasets() {
-        if( workspaceId )
-            return "$endpoint/workspaces/$workspaceId/datasets/"
-        return "$endpoint/datasets/"
-    }
-
-    protected String getUrlDatasetUpload(String datasetId) {
-        if( workspaceId )
-            return "$endpoint/workspaces/$workspaceId/datasets/$datasetId/upload"
-        return "$endpoint/datasets/$datasetId/upload"
-    }
-
     /**
      * On workflow start, submit a message with some basic
      * information, like Id, activity and an ISO 8601 formatted
@@ -959,7 +979,7 @@ class TowerClient implements TraceObserverV2 {
     }
 
     /**
-     * Create a new dataset in Seqera Platform
+     * Create a new dataset in Seqera Platform using tower-java-sdk
      *
      * @param name The name for the new dataset
      * @param description The description for the new dataset
@@ -968,40 +988,39 @@ class TowerClient implements TraceObserverV2 {
     private String createDataset(String name, String description) {
         log.info "Creating new dataset: ${name}"
 
+        if( !datasetsApi ) {
+            log.warn "DatasetsApi not initialized, cannot create dataset"
+            return null
+        }
+
         try {
-            final payload = [
-                name: name,
-                description: "Workflow output: ${description}",
-                header: true
-            ]
-
-            final url = getUrlDatasets()
-            final resp = sendHttpMessage(url, payload, 'POST')
-
-            if( resp.isError() ) {
-                log.warn "Failed to create dataset '${name}': ${resp.message}"
-                return null
-            }
+            def request = new CreateDatasetRequest()
+            request.setName(name)
+            request.setDescription("Workflow output: ${description}")
 
-            // Parse the response to extract dataset ID
-            final json = new JsonSlurper().parseText(resp.message) as Map
-            final dataset = json.dataset as Map
-            final datasetId = dataset?.id as String
+            def wspId = workspaceId ? Long.valueOf(workspaceId) : null
+            CreateDatasetResponse response = datasetsApi.createDataset(wspId, request)
+
+            def datasetId = response.dataset?.id?.toString()
 
             if( datasetId ) {
                 log.info "Created dataset '${name}' with ID: ${datasetId}"
             }
 
             return datasetId
         }
+        catch( ApiException e ) {
+            log.warn "Failed to create dataset '${name}': ${e.message} (status: ${e.code})", e
+            return null
+        }
         catch( Exception e ) {
             log.warn "Failed to create dataset '${name}': ${e.message}", e
             return null
         }
     }
 
     /**
-     * Upload an index file to a dataset
+     * Upload an index file to a dataset using tower-java-sdk
      *
      * @param datasetId The ID of the dataset
      * @param indexPath The path to the index file
@@ -1013,96 +1032,27 @@ class TowerClient implements TraceObserverV2 {
             return
         }
 
-        log.info "Uploading index file for output '${outputName}' to dataset ${datasetId}: ${indexPath}"
-
-        try {
-            // Build URL with header parameter
-            def url = getUrlDatasetUpload(datasetId)
-            // Workflow output index files always have headers
-            url += "?header=true"
-
-            // Upload file using multipart form data
-            final resp = uploadFile(url, indexPath.toFile())
-
-            if( resp.isError() ) {
-                log.warn "Failed to upload index file for output '${outputName}': ${resp.message}"
-            } else {
-                log.info "Successfully uploaded index file for output '${outputName}' to dataset ${datasetId}"
-            }
-        }
-        catch( Exception e ) {
-            log.warn "Failed to upload index file for output '${outputName}': ${e.message}", e
+        if( !datasetsApi ) {
+            log.warn "DatasetsApi not initialized, cannot upload index file"
+            return
         }
-    }
 
-    /**
-     * Upload a file to Seqera Platform using multipart/form-data
-     *
-     * @param url The upload URL
-     * @param file The file to upload
-     * @return Response object
-     */
-    protected Response uploadFile(String url, File file) {
-        log.trace "HTTP multipart upload: url=$url; file=${file.name}"
+        log.info "Uploading index file for output '${outputName}' to dataset ${datasetId}: ${indexPath}"
 
         try {
-            // Create multipart body
-            final boundary = "----TowerNextflowBoundary" + System.currentTimeMillis()
-            final body = createMultipartBody(file, boundary)
-
-            // Build request
-            final request = HttpRequest.newBuilder(URI.create(url))
-                .header('Content-Type', "multipart/form-data; boundary=$boundary")
-                .header('User-Agent', "Nextflow/$BuildInfo.version")
-                .header('Traceparent', TraceUtils.rndTrace())
-                .POST(HttpRequest.BodyPublishers.ofByteArray(body))
-                .build()
-
-            final resp = httpClient.sendAsString(request)
-            final status = resp.statusCode()
+            def wspId = workspaceId ? Long.valueOf(workspaceId) : null
+            def header = Boolean.TRUE  // Workflow output index files always have headers
 
-            if( status == 401 ) {
-                return new Response(status, 'Unauthorized Seqera Platform API access')
-            }
-            if( status >= 400 ) {
-                final msg = parseCause(resp?.body()) ?: "Unexpected response for request $url"
-                return new Response(status, msg as String)
-            }
+            datasetsApi.uploadDataset(wspId, datasetId, header, indexPath.toFile())
 
-            return new Response(status, resp.body())
+            log.info "Successfully uploaded index file for output '${outputName}' to dataset ${datasetId}"
         }
-        catch( IOException e ) {
-            return new Response(0, "Unable to connect to Seqera Platform API: ${getHostUrl(url)}")
+        catch( ApiException e ) {
+            log.warn "Failed to upload index file for output '${outputName}': ${e.message} (status: ${e.code})", e
+        }
+        catch( Exception e ) {
+            log.warn "Failed to upload index file for output '${outputName}': ${e.message}", e
         }
-    }
-
-    /**
-     * Create a multipart/form-data request body
-     *
-     * @param file The file to include in the request
-     * @param boundary The multipart boundary string
-     * @return Byte array containing the multipart body
-     */
-    private byte[] createMultipartBody(File file, String boundary) {
-        final baos = new ByteArrayOutputStream()
-        final writer = new PrintWriter(new OutputStreamWriter(baos, 'UTF-8'), true)
-
-        // Write file part
-        writer.append("--${boundary}\r\n")
-        writer.append("Content-Disposition: form-data; name=\"file\"; filename=\"${file.name}\"\r\n")
-        writer.append("Content-Type: text/csv\r\n")
-        writer.append("\r\n")
-        writer.flush()
-
-        // Write file content
-        baos.write(file.bytes)
-
-        // Write closing boundary
-        writer.append("\r\n")
-        writer.append("--${boundary}--\r\n")
-        writer.flush()
-
-        return baos.toByteArray()
     }
 
 }