Ingestion Options

To ingest data into Flowcore, you can see the available options by clicking on the “Ingestion Channels” page. It will show you the available options for ingestion.

Webhook

The simplest and fastest way to get data into Flowcore is to use webhooks. You can use the webhook generation feature to create a url that you can use to send data to Flowcore.

Clicking on the “integrate” link will take you to the webhook generation page where you can create a webhook.

Then select the Data Core, Flow Type and Event Type you want to send data to. Fill in the API key and you will be presented with two URLs. One for sending single events and the other for sending multiple events.

You can simply post your event data as JSON to the URL and it will be ingested into Flowcore, into the selected location.

{
  // event data
}

Metadata

You can also add metadata to the event by adding a base64 encoded stringified JSON value x-flowcore-metadata-json header when sending the event.

{
  "x-flowcore-metadata-json": "eyJtZXRhZGF0YSI6Im1ldGFkYXRhIn0="
}

Hot Stroage TTL

You can customize how long the event is stored in hot storage before it is deleted by setting ttl-on/stored-event to true in the metadata object. This accepts d, h, m, s. The maximum value is 7 days, this can be increased for higher subscription levels.

Ephemeral events

You can also opt out from storing the event in cold storage by setting the do-not-archive-on/stored-event to true. This will make the event available for the ttl specified, after which it will be deleted.

Bi-Temporality

Events in Flowcore are stored using bi-temporality. This means that each event has two timestamps associated with it. One is the event time timestamp, which is the time the event was ingested into Flowcore. The other is the valid time timestamp, which is the time the event is valid from.

The default valid time is the event time, and the event time is the time the event was ingested into Flowcore.

You can override both these values by using a set of headers at ingestion time.

You can override the valid time by using the x-flowcore-valid-time Headers, and pass in an ISO 8601 timestamp.

You can override the event time by using the x-flowcore-event-time Headers, and pass in an ISO 8601 timestamp.

if you are sending a batch of events, you can override the event time or valid time by specifying which field in the data you want to use as the event time or valid time. with the use of the x-flowcore-event-time-key and x-flowcore-valid-time-key headers respectively. Values for the fields are in the same format as the headers above.

Multible Events

If you want to send multiple events at once, you can use the batch URL. This will allow you to send multiple events in a single request.

The payload should be an array of events.

[
  {
    // event 1
  },
  {
    // event 2
  }
]

Multipart File Events

It is also possible to send files and have them be converted to multipart events. This is useful if you want to also store files and images in Flowcore.

This uses the same URL as the batch URL, but the payload should be a multipart form data.

{
  "file": "file",
  "type": "image/png",
}

{
  "fileName": "file.png",
  "fileSize": 1024,
  "fileType": "image/png",
  "part": 1,
  "totalParts": 2,
  "data": "base64 encoded data buffer",
  "checksum": "SHA256 checksum",
  "hashType": "SHA256"
}

Multipart Reconstruction

To reconstruct the file, you simply need to get the events that are part of the file, and then concat the data together. You can compare the checksum of the reconstructed file with the original file to ensure that the file is intact.

const sorted = events.sort((a, b) => a.part - b.part);

const data = sorted.map((event) => Buffer.from(event.data, "base64"));

const payload = Buffer.concat(data);

fs.writeFileSync("reconstructed-file.png", payload);

One way you can use this for example in a database is to have a write queue where all events are stored, then a service/worker that periodically scans the queue for completed files, reconstructs them and stores them in a database. Then after the queue can be cleared for the reconstructed file.