Connect Clickhouse and Elasticsearch in our serverless environment

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

    async fetch(publish, done) {
        // TODO: customize your sql query
        const sql = "SELECT * FROM your_table_name";

        const resultSet = await this.clickhouse.query({
            query: sql,
            format: "CSV",
        });
        const stream = resultSet.stream();
        stream.on("data", (rows) => {
            rows.forEach(publish);
        });
        stream.on("end", done);
    }

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

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

    async consume(item) {
        // TODO: Map your item to row
        await this.elasticsearch.index({
            index: "your_table_name",
            document: item,
        });
    }

    async close() {}
}