package io.digdag.standards.operator.td;

import com.google.inject.Inject;
import com.treasuredata.client.TDClientException;
import com.treasuredata.client.model.TDExportResultJobRequest;
import io.digdag.client.config.Config;
import io.digdag.core.Environment;
import io.digdag.spi.Operator;
import io.digdag.spi.OperatorContext;
import io.digdag.spi.OperatorFactory;
import io.digdag.spi.TaskResult;
import io.digdag.util.UserSecretTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class TDResultExportOperatorFactory
        implements OperatorFactory
{
    private final Map<String, String> env;
    private final Config systemConfig;
    private static Logger logger = LoggerFactory.getLogger(TdTableExportOperatorFactory.class);
    private final BaseTDClientFactory clientFactory;

    @Inject
    public TDResultExportOperatorFactory(@Environment Map<String, String> env, Config systemConfig, BaseTDClientFactory clientFactory)
    {
        this.env = env;
        this.systemConfig = systemConfig;
        this.clientFactory = clientFactory;
    }

    @Override
    public String getType() { return "td_result_export"; }

    @Override
    public Operator newOperator(OperatorContext context) { return new TDResultExportOperator(context, clientFactory); }

    private class TDResultExportOperator
            extends BaseTdJobOperator
    {
        private final String jobId;
        private final String resultSettings;
        private final String resultConnection;

        private TDResultExportOperator(OperatorContext context, BaseTDClientFactory clientFactory)
        {
        super(context, env, systemConfig, clientFactory);
        Config params = request.getConfig();
        this.jobId = params.get("job_id", String.class);
        this.resultConnection = params.get("result_connection", String.class);

        UserSecretTemplate template = UserSecretTemplate.of(params.parseNested("result_settings").toString());
        this.resultSettings = template.format(context.getSecrets());
    }

        @Override
        protected String startJob(TDOperator op, String domainKey)
        {
            TDExportResultJobRequest jobRequest = TDExportResultJobRequest.builder()
                    .jobId(this.jobId)
                    .resultConnectionId(String.valueOf(op.lookupConnection(this.resultConnection)))
                    .resultConnectionSettings(this.resultSettings)
                    .build();

            String jobId = op.submitNewJobWithRetry(client -> client.submitResultExportJob(jobRequest));
            logger.info("Started result export job id={}", jobId);

            return jobId;
        }

        @Override
        public final TaskResult runTask()
        {
            try (TDOperator op = TDOperator.fromConfig(clientFactory, systemDefaultConfig, env, params, context.getSecrets().getSecrets("td"), false)) {
                return runTask(op);
            }
            catch (TDClientException ex) {
                throw propagateTDClientException(ex);
            }
        }
    }
}

@@ -1640,16 +1640,20 @@
         
+    
 super(co
@@ -1702,16 +1702,20 @@
         
+   
+ 
 Config p
@@ -1743,16 +1743,20 @@
 nfig();%0A
+    
         
@@ -1804,32 +1804,36 @@
 class);%0A        
+  
+  
 this.resultConne
@@ -1884,24 +1884,28 @@
 ng.class);%0A%0A
+    
         User
@@ -2003,32 +2003,36 @@
 ing());%0A        
+ 
+   
 this.resultSetti
@@ -2072,24 +2072,28 @@
 Secrets());%0A
+    
     %7D%0A%0A     

Fix indent