Batch Upload: Bulk Event Ingestion

{"metric_id":"api_calls","quantity":1,"tenant_id":"tenant_acme","event_time":"2026-04-15T10:22:00Z","idempotency_key":"evt_acme_1001","customer_id":"cust_alice","properties":{"endpoint":"/v1/search","method":"POST"}}
{"metric_id":"api_calls","quantity":1,"tenant_id":"tenant_acme","event_time":"2026-04-15T10:22:04Z","idempotency_key":"evt_acme_1002","customer_id":"cust_alice","properties":{"endpoint":"/v1/search","method":"POST"}}
{"metric_id":"ai_tokens","quantity":3840,"tenant_id":"tenant_globex","event_time":"2026-04-15T10:23:11Z","idempotency_key":"evt_globex_0091","customer_id":"cust_bob","properties":{"model":"claude-opus-4","type":"input"}}
{"metric_id":"storage_gb","quantity":0.42,"tenant_id":"tenant_initech","event_time":"2026-04-15T10:24:00Z","idempotency_key":"evt_initech_0055"}

curl -X POST https://api.aforo.ai/v1/ingest/batch \
  -H "Authorization: Bearer $AFORO_API_KEY" \
  -H "X-Tenant-Id: tenant_acme" \
  -F "file=@usage-events.ndjson;type=application/x-ndjson" \
  -F "dry_run=false" \
  -F "notify_email=ops@acme.com"

# 202 Accepted response
{
  "job_id":     "job_01HXKRM3ABCDE7F8G9HNPQRST",
  "status":     "QUEUED",
  "file_name":  "usage-events.ndjson",
  "file_size_bytes": 12847,
  "estimated_events": 10000,
  "created_at": "2026-04-18T09:22:00Z",
  "status_url": "https://api.aforo.ai/v1/ingest/batch/job_01HXKRM3ABCDE7F8G9HNPQRST"
}

curl https://api.aforo.ai/v1/ingest/batch/job_01HXKRM3ABCDE7F8G9HNPQRST \
  -H "Authorization: Bearer $AFORO_API_KEY" \
  -H "X-Tenant-Id: tenant_acme"

# Processing response (job still running)
{
  "job_id":           "job_01HXKRM3ABCDE7F8G9HNPQRST",
  "status":           "PROCESSING",
  "progress_pct":     62,
  "events_accepted":  6184,
  "events_rejected":  3,
  "events_remaining": 3813,
  "started_at":       "2026-04-18T09:22:05Z"
}

# Completed response
{
  "job_id":          "job_01HXKRM3ABCDE7F8G9HNPQRST",
  "status":          "COMPLETED",
  "events_total":    10000,
  "events_accepted": 9994,
  "events_rejected": 6,
  "events_dupes":    0,
  "started_at":      "2026-04-18T09:22:05Z",
  "completed_at":    "2026-04-18T09:24:11Z",
  "error_file_url":  "https://api.aforo.ai/v1/ingest/batch/job_.../errors.ndjson"
}

{"line":7,"error_code":"INVALID_METRIC_ID","error_message":"metric_id 'api_requests' not found. Did you mean 'api_calls'?","original":{"metric_id":"api_requests","quantity":1,"tenant_id":"tenant_acme","event_time":"2026-04-15T10:22:00Z"}}
{"line":23,"error_code":"TIMESTAMP_TOO_OLD","error_message":"event_time '2025-12-01T00:00:00Z' is 139 days in the past. Maximum lookback is 90 days.","original":{"metric_id":"api_calls","quantity":1,"tenant_id":"tenant_acme","event_time":"2025-12-01T00:00:00Z"}}
{"line":91,"error_code":"QUANTITY_NOT_POSITIVE","error_message":"quantity must be > 0, got: -5","original":{"metric_id":"storage_gb","quantity":-5,"tenant_id":"tenant_acme","event_time":"2026-04-15T10:44:00Z"}}

#!/usr/bin/env node
// Usage: node csv-to-ndjson.js input.csv output.ndjson

import { createReadStream, createWriteStream } from 'fs';
import { createInterface }                      from 'readline';
import { randomUUID }                           from 'crypto';

const [,, inputFile, outputFile] = process.argv;
if (!inputFile || !outputFile) {
  console.error('Usage: node csv-to-ndjson.js input.csv output.ndjson');
  process.exit(1);
}

// ── Column mapping: adjust to match your CSV headers ──────────────────────────
const COLUMN_MAP = {
  metric_id:        'metric',         // CSV column → Aforo field
  quantity:         'count',
  tenant_id:        'account_id',
  customer_id:      'user_id',
  event_time:       'occurred_at',
};
// ──────────────────────────────────────────────────────────────────────────────

const rl      = createInterface({ input: createReadStream(inputFile) });
const out     = createWriteStream(outputFile);
let   headers = null;
let   lineNo  = 0;
let   accepted = 0;
let   skipped  = 0;

rl.on('line', (line) => {
  lineNo++;
  if (lineNo === 1) {
    headers = line.split(',').map(h => h.trim());
    return;
  }

  const cols   = line.split(',');
  const record = Object.fromEntries(headers.map((h, i) => [h, cols[i]?.trim() ?? '']));

  const event = {
    metric_id:        record[COLUMN_MAP.metric_id],
    quantity:         parseFloat(record[COLUMN_MAP.quantity]),
    tenant_id:        record[COLUMN_MAP.tenant_id],
    customer_id:      record[COLUMN_MAP.customer_id] || undefined,
    event_time:       new Date(record[COLUMN_MAP.event_time]).toISOString(),
    idempotency_key:  `backfill-${inputFile}-${lineNo}`,
  };

  // Basic validation before writing
  if (!event.metric_id || isNaN(event.quantity) || !event.tenant_id || event.event_time === 'Invalid Date') {
    console.warn(`Line ${lineNo}: skipped (missing required fields)`);
    skipped++;
    return;
  }

  out.write(JSON.stringify(event) + '\n');
  accepted++;
});

rl.on('close', () => {
  out.end();
  console.log(`Done. ${accepted} events written, ${skipped} skipped.`);
  console.log(`Output: ${outputFile}`);
});

# Convert and validate before uploading
node csv-to-ndjson.js legacy-usage-export.csv usage-events.ndjson

# Dry run first — validates without ingesting
curl -X POST https://api.aforo.ai/v1/ingest/batch \
  -H "Authorization: Bearer $AFORO_API_KEY" \
  -H "X-Tenant-Id: tenant_acme" \
  -F "file=@usage-events.ndjson;type=application/x-ndjson" \
  -F "dry_run=true"

# If dry_run shows 0 rejected events, submit for real
curl -X POST https://api.aforo.ai/v1/ingest/batch \
  -H "Authorization: Bearer $AFORO_API_KEY" \
  -H "X-Tenant-Id: tenant_acme" \
  -F "file=@usage-events.ndjson;type=application/x-ndjson" \
  -F "notify_email=ops@acme.com"