Pipelines

To create sources and destinations, refer to the @event-inc/connections package.

import { 
  createClient,
  createPipeline,
  updatePipeline,
  deletePipeline
} from "@event-inc/pipelines";

const main = async () => {
  const client = createClient(process.env.EVENT_INC_TOKEN, {
    environment: "sandbox" // or "production"
  });

  // Create a pipeline
  const pipeline = await createPipeline(client, {
    label: `Stripe to Elasticsearch`,
    source: {
      key: "SOURCE_KEY",
      events: ["customer.created"], // Events to trigger pipeline
    },
    destination: {
      key: "DESTINATION_KEY",
      action: "index", // Destination action to trigger
    },

    // Extract data to enrich transformation context
    extractors: [
      {
        type: "http",
        label: "customer",
        config: {
          url: "https://api.yourcompanydomain.co",
          method: "GET"
        },
      },
    ],
    transformation: {
      type: "javascript",

      // Transform data before sending it to the destination
      func: function transform(event, context) {
        const body = JSON.parse(event.body);
        // const extractorData = context?.extractorLabel?.data;

        return {
          index: "stripe-customers",
          document: {
            name: body.data.object.email,
          },
        };
      },
    },
    // Pipeline configuration
    settings: {
      startToCloseTimeout: "30 minutes", // Execution window
      retry: {
        maximumAttempts: 10, // Maximum retry attempts within window
      },
    },
  });

  // Update the pipeline
  const updatedPipeline = await updatePipeline(client, {
    key: "SOURCE_KEY",
    active: false
  });

  // Delete the pipeline
  await deletePipeline(client, {
    key: pipeline.key,
  });
}

main();