import * as blob from "@azure/storage-blob";
import { from } from "@data-heaving/orchestration";
import * as mssqlSource from "@data-heaving/source-sql-mssql";
import csvTransform from "@data-heaving/transform-csv";
import gzipTransform from "@data-heaving/transform-gzip";
import * as blobSink from "@data-heaving/sink-azure-blob";
import * as azureCommon from "@data-heaving/common-azure";
import * as events from "./events";

// Prepare pipeline elements
const sqlPool = mssqlSource.getMSSQLPool(...); // Pool of connections to SQL server, this example will only use 1 connection though
const auth = azureCommon.getEnvOrManagedIDAuth(); // Azure Authentication functionality
const ctStorage = new blob.ContainerClient('url-to-storage-container', auth); // Where to store change tracking information for the table
const eventBuilder = events.createMyEventBuilder(); // See below for more info
// Set up event builder with listeners here, before
const eventEmitter = eventBuilder.createEventEmitter();

// Define the pipeline
const pipeline = from(
  mssqlSource.rowsInTable(sqlPool)
    .fullLoad(eventEmitter)
  ))
  // Transform SQL rows to CSV rows (strings), without header as we don't get table metadata as context
  .simpleTransformEveryDatum(
    csvTransform(),
  )
  // Compress CSV rows (strings) using GZIP
  .complexTransformEveryDatum(gzipTransform())
  // And store them to Azure BLOB storage, splitting to approximately 150MB size each (as optimal for Snowflake ingestion)
  .storeTo(
    blobSink.toAzureBlobStorage({
      getBlobID: ({ tableID, tableProcessingStartTime }) =>
      // Construct folder path by using just table name + load time
        `${azureCommon.sanitizeForBlobPath(
          tableID.tableName,
        )}/${tableProcessingStartTime}`,
      blobClientFactory: (blobID, existingCount) => ({
        maxSizeInKB: 150 * 1024, // Approximate max single compressed file size: 150MB
        client: dataClient.getBlockBlobClient(
          `${blobID}/data-${existingCount}.csv.gz`, // Individual files within the folder will be data-0.csv.gz, data-1.csv.gz, etc.
        ),
      }),
      eventEmitter,
    }),
  )
  .finalizePipeline();

// Invoke the pipeline
await pipeline({
  databaseName: 'myDB',
  schemaName: 'mySchema',
  tableName: 'myTable'
});