Migrate files, databases, and object storage

GraphJSON does not require the source to be another analytics product. Any system that can produce timestamped JSON events can migrate through POST /api/bulk-log.

This guide covers:

• CSV
• newline-delimited JSON
• PostgreSQL or another relational database
• warehouse query results
• object-storage partitions

Use a staged workflow. Do not connect an unrestricted production database directly to an unbounded importer.

Define the target history#

Start with reports that must survive:

Report:
Required event:
Required properties:
Identity grain:
Historical start:
Historical end:
Time zone:
Source authority:

Import only history that supports a real decision, comparison, cohort, customer context, or obligation.

Old files and tables often contain:

• obsolete schemas
• duplicate rows
• soft-deleted test records
• direct identifiers
• free text
• internal operational columns

Do not preserve them merely because they exist.

Choose event facts, not current rows#

A current application table usually represents state:

accounts
  id
  current_plan
  current_status
  updated_at

It cannot reconstruct every historical plan change unless an audit or change table exists.

Good migration sources:

• append-only event table
• audit/change table
• order or invoice ledger
• immutable object-store events
• provider export with occurrence timestamps

If only current state exists, import an explicit snapshot:

{
  "event": "account_state_snapshot",
  "account_id": "acct_7",
  "status": "active",
  "plan": "pro",
  "snapshot_at": 1785081600,
  "source": "accounts_table_2026_07"
}

Do not label a snapshot as the historical event that originally created the state.

Create a migration manifest#

{
  "migration_id": "mig_orders_2026_07_v1",
  "source_type": "postgresql",
  "source_snapshot": "orders_export_2026_07_26",
  "target_collection": "order_events",
  "transform_version": "order-event-v2",
  "start_time": 1735689600,
  "end_time": 1785081599,
  "expected_source_rows": 984120
}

Record:

• source query or export command
• immutable file checksum or database snapshot identifier
• inclusion and exclusion rules
• source time zone
• target schema
• expected rows
• owner

The manifest is the migration control record. Keep it outside GraphJSON.

Normalize into one import row#

type ImportRow = {
  source_position: string;
  event_id: string;
  occurred_at: number;
  collection: string;
  payload: Record<string, unknown>;
};

Every source adapter should produce this type before delivery.

This lets CSV, database, and object-storage migrations share the same:

• schema validator
• privacy checks
• batch sender
• checkpoint store
• dead-letter workflow
• reconciliation

Import CSV safely#

Use a real CSV parser. Do not split lines or commas manually; quoted values, embedded newlines, encodings, and escaped quotes make that unsafe.

Expected source:

order_id,account_id,completed_at,amount_minor,currency
ord_1,acct_7,2026-07-25T18:12:00Z,4900,usd

Transform:

function orderFromCsv(
  row: Record<string, string>,
  line: number
): ImportRow {
  const occurredAt = Math.floor(
    new Date(row.completed_at).getTime() / 1000
  );

  if (!Number.isInteger(occurredAt)) {
    throw new Error("invalid_completed_at");
  }

  return {
    source_position: `orders.csv:${line}`,
    event_id: `order:${row.order_id}:completed`,
    occurred_at: occurredAt,
    collection: "order_events",
    payload: {
      event: "order_completed",
      event_id: `order:${row.order_id}:completed`,
      account_id: row.account_id,
      order_id: row.order_id,
      amount_minor: Number(row.amount_minor),
      currency: row.currency.toLowerCase(),
      schema_version: 1
    }
  };
}

Reject:

• invalid time
• missing ID
• non-integer money
• unsupported currency
• duplicate source identity

Do not coerce a malformed numeric string to zero.

Import NDJSON#

NDJSON stores one JSON object per line:

{"event":"job_completed","id":"job_1","completed_at":1785081600}
{"event":"job_completed","id":"job_2","completed_at":1785081610}

It is well suited to:

• streaming
• line checkpoints
• independent rejected rows
• compressed object-store partitions

Process line-by-line rather than loading a multi-gigabyte file into memory.

For each line:

1. record file and line position
2. parse one JSON object
3. validate source schema
4. map to the target event schema
5. write accepted or quarantined output

A blank line can be skipped deliberately. A malformed line should not cause the importer to lose the checkpoint for every prior acknowledged batch.

Extract from PostgreSQL#

Use:

• read-only credentials
• a replica or controlled snapshot when possible
• explicit selected columns
• keyset pagination
• a stable order

Example extraction query:

SELECT
  id,
  account_id,
  completed_at,
  amount_minor,
  currency
FROM orders
WHERE
  status = 'completed'
  AND completed_at >= $1
  AND completed_at < $2
  AND id > $3
ORDER BY id
LIMIT 10000;

Checkpoint the last exported id.

Do not use increasing OFFSET for a large mutable table. It becomes slower and can skip or repeat rows when the source changes.

If IDs do not order with event time, the extraction can still checkpoint by ID while preserving completed_at as the GraphJSON occurrence timestamp.

For a consistent historical snapshot, record the snapshot boundary and ensure the source’s mutation policy cannot change earlier exported pages unnoticed.

Extract state changes#

When a relational table contains updates rather than events, prefer its audit log:

SELECT
  change_id,
  account_id,
  previous_status,
  new_status,
  changed_at
FROM account_status_changes
WHERE change_id > $1
ORDER BY change_id
LIMIT 10000;

Map each row to:

{
  "event": "account_status_changed",
  "change_id": "chg_71",
  "account_id": "acct_7",
  "previous_status": "trialing",
  "status": "active",
  "schema_version": 1
}

Do not generate a historical change timeline by comparing only today’s table with a guess.

Extract from a warehouse#

A warehouse query should return event-grain rows:

SELECT
  event_id,
  occurred_at,
  event_name,
  user_id,
  account_id,
  schema_version
FROM analytics.cleaned_product_events
WHERE occurred_at >= :start
  AND occurred_at < :end
ORDER BY occurred_at, event_id;

Record:

• query text or version
• warehouse snapshot/table version
• selected time interval
• row count
• result checksum by partition when practical

Do not export a dashboard aggregate and relabel each total as a raw product event.

Import object-storage partitions#

Recommended layout:

events/
  date=2026-07-24/part-000.ndjson.gz
  date=2026-07-24/part-001.ndjson.gz
  date=2026-07-25/part-000.ndjson.gz

Manifest each object:

object key
object version
etag or checksum
compressed bytes
expected rows
status
last acknowledged line

Process a deterministic object order and checkpoint within each object.

Do not assume an object listing is an immutable snapshot. Record exact object versions when the storage system supports versioning.

Decompress as a stream. Keep decompression failures separate from JSON-schema rejections.

Normalize time once#

GraphJSON requires whole Unix seconds.

Source values may be:

• ISO 8601 with offset
• timestamp without time zone
• Unix seconds
• Unix milliseconds
• database-native timestamp

Make the source policy explicit:

function unixSecondsFromMilliseconds(value: number) {
  if (!Number.isSafeInteger(value) || value < 1_000_000_000_000) {
    throw new Error("expected_unix_milliseconds");
  }
  return Math.floor(value / 1000);
}

Do not use a numeric magnitude heuristic without a source-specific contract.

Preserve the original occurrence time. Import time belongs in migration metadata, not the GraphJSON request timestamp.

Derive stable event IDs#

Prefer an existing immutable source event ID.

If none exists, derive from a stable tuple:

event_id =
  "order:" + order_id + ":completed:" + source_version

For a hash:

• define canonical field order
• define separators and encoding
• include an algorithm version
• test collisions
• never include active secrets

Do not derive an ID from line number alone when the source can be repartitioned.

Deliver through the bulk API#

async function sendBatch(rows: ImportRow[]) {
  const collection = rows[0].collection;

  if (!rows.every((row) => row.collection === collection)) {
    throw new Error("mixed_collection_batch");
  }

  const response = await fetch(
    "https://api.graphjson.com/api/bulk-log",
    {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        api_key: process.env.GRAPHJSON_API_KEY,
        collection,
        timestamps: rows.map((row) => row.occurred_at),
        jsons: rows.map((row) => JSON.stringify(row.payload))
      })
    }
  );

  const body = await response.text();
  if (!response.ok) {
    throw new Error(`GraphJSON ${response.status}: ${body.slice(0, 300)}`);
  }
}

Keep the key in a restricted migration environment. Do not put it inside staged files.

Use High-volume ingestion and backfills for concurrency, retries, checkpoints, and abort conditions.

Pilot the difficult partition#

Choose a small partition containing:

• oldest supported schema
• newest schema
• missing optional values
• largest payload
• duplicate candidate
• time-zone boundary
• non-ASCII string

Validate:

• Samples
• types
• first and last occurrence time
• counts by event
• identity
• intended SQL and dashboards

Reconcile#

Compare:

source rows selected
- intentionally excluded
- quarantined
= expected GraphJSON facts

Then compare by:

• source partition
• day
• event name
• schema version
• account/user presence
• financial or operational totals

Vendor or warehouse report totals may apply hidden identity, time-zone, or filter behavior. Reconcile raw facts first.

Cut over#

For a new production producer:

1. stabilize the target contract
2. start new event delivery
3. import historical data separately
4. verify overlap without double counting
5. switch dashboards
6. monitor one full reporting cycle
7. retire the old export path deliberately

Keep the migration transform and manifest long enough to explain historical differences.

Migration checklist#

• Required history is tied to real reports.
• Source represents event facts or an explicit snapshot.
• Manifest records immutable input and transform.
• Read access is scoped and read-only.
• CSV uses a real parser.
• NDJSON and objects stream with checkpoints.
• Database extraction uses stable keyset pagination.
• Time normalization is source-specific.
• Stable event IDs survive retries.
• Unsafe or malformed rows are quarantined.
• Bulk delivery is bounded and restartable.
• A difficult pilot partition passed.
• Raw, destination, and business totals reconcile.

Continue with Migration overview, Bulk logging, and Executable event contracts.