# Ingest events

[Read this page in HTML](https://www.scopedb.io/docs/guides/ingest-events)

Start with a small batch you can inspect, then connect the producer that will keep sending events. This guide uses the `events` table from the [quickstart](https://www.scopedb.io/docs/guides/quickstart.md).

Run the ScopeQL blocks in [Query](https://www.scopedb.io/docs/guides/console-query.md), or save each block to a file and use `scope query -f <file>` after [connecting your client](https://www.scopedb.io/docs/guides/connect-to-scopedb.md).

## Write a small batch

The destination has `time`, `service`, `name`, `message`, and `var` columns. For events that already have this shape, insert a small batch directly:

```scopeql
VALUES (
    NOW(),
    'checkout',
    'payment_failed',
    'Credit card gateway timed out',
    {'region': 'apac', 'gateway': 'credit_card'}
)
SELECT
    $0 AS time,
    $1::string AS service,
    $2::string AS name,
    $3::string AS message,
    $4::object AS var
INSERT INTO events
```

Run the insert once. Repeating it adds another event.

## Keep the original payload when reshaping events

When producers send different JSON shapes, extract the fields used by common queries and keep the original object in `var`. This sample turns one source payload into an event row:

```scopeql
VALUES ({
    'service': 'checkout',
    'name': 'payment_failed',
    'message': 'Credit card gateway timed out',
    'gateway': 'credit_card'
})
SELECT
    NOW() AS time,
    $0['service']::string AS service,
    $0['name']::string AS name,
    $0['message']::string AS message,
    $0::object AS var
WHERE service IS NOT NULL
  AND name IS NOT NULL
INSERT INTO events
```

This example uses arrival time. When source events have their own timestamp, preserve that event time so delayed deliveries appear in the correct investigation window. Agree on required fields and how to handle missing values before connecting a production writer; see [model event data](https://www.scopedb.io/docs/guides/model-flexible-events.md).

## Verify the new records

Query a recent window and inspect the fields you just sent:

```scopeql
FROM events
WHERE time >= NOW() - 'PT15m'::interval
  AND name = 'payment_failed'
SELECT time, service, message, var['gateway']::string AS gateway
ORDER BY time DESC
LIMIT 10
```

Look for `checkout`, the timeout message, and `credit_card`. You can also open `events` in [Explore](https://www.scopedb.io/docs/guides/console-explore.md) and use the same time range and filter. If the data is missing, check the destination workspace and table, source timestamps, and write result before sending the batch again.

## Keep writing from your application

For a continuous producer whose events already match the table schema, use an SDK writer to batch rows: [JavaScript](https://www.scopedb.io/docs/guides/sdks/nodejs.md#write-rows), [Go](https://www.scopedb.io/docs/guides/sdks/go.md#write-rows), or [Rust](https://www.scopedb.io/docs/guides/sdks/rust.md#write-rows). For other runtimes, use the [HTTP write guide](https://www.scopedb.io/docs/api-reference/append.md). When application events need a server-side ScopeQL transformation, append the raw JSON payloads into an `object` column in a staging table and run a `FROM ... SELECT ... INSERT INTO` statement, or submit small `VALUES ... SELECT ... INSERT INTO` batches through your SDK or the [HTTP statements API](https://www.scopedb.io/docs/api-reference/statements.md).

Check the client's delivery result before retrying a failed write. A connection failure can leave the outcome unknown; repeating the batch can duplicate events. Use a stable event identifier when your application needs to recognize repeated deliveries.

Continue with [query events](https://www.scopedb.io/docs/guides/query-events.md) to investigate failures and compare event counts.

## Related contents

* [Quickstart](https://www.scopedb.io/docs/guides/quickstart.md)

* [Query events](https://www.scopedb.io/docs/guides/query-events.md)

* [Append rows over HTTP](https://www.scopedb.io/docs/api-reference/append.md)