Connect Amazon S3 NDJSON file and Microsoft SQL Server in our serverless environment

class AwsS3SourceReadRemoteNdjson {
    async init() {
        // TODO: Create your aws-s3 credential
        // More info at https://yepcode.io/docs/integrations/aws-s3/#credential-configuration
        this.awsS3 = yepcode.integration.awsS3("your-aws-s3-credential-name");
    }

    async fetch(publish, done) {
        // TODO: Customize the command to be executed
        // More info at: https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-s3/interfaces/getobjectcommandinput.html
        const getObjectCommand = new GetObjectCommand({
            Bucket: "your-bucket-name",
            Key: "your-file-name",
        });

        const getObjectResult = await this.awsS3.send(getObjectCommand);

        getObjectResult.body
            .pipe(ndjson.parse())
            .on("data", async (line) => {
                await publish(line);
            })
            .on("end", done);
    }

    async close() {}
}

class MssqlTargetInsert {
    async init() {
        // TODO: Create your mssql credential
        // More info at https://yepcode.io/docs/integrations/mssql/#credential-configuration
        const mssqlConnectionPool = yepcode.integration.mssql(
            "your-mssql-credential-name"
        );
        this.pool = await mssqlConnectionPool.connect();
    }

    async consume(item) {
        // TODO: Customize input params and INSERT statement
        await this.pool
            .request()
            .input("name", mssql.VarChar(50), item.name)
            .input("price", mssql.Int, item.price)
            .input("stock", mssql.Int, item.stock)
            .query(
                "INSERT INTO products(name, price, stock, created_at) VALUES(@name, @price, @stock, CURRENT_TIMESTAMP)"
            );
    }

    async close() {
        await this.pool.close();
    }
}