DataSync Event Ingestion

A production-ready data ingestion system that extracts ~3M events from the DataSync Analytics API and stores them in PostgreSQL, using a Producer-Consumer architecture with Redis Streams as the message buffer.

How to Run

sh run-ingestion.sh

This single command builds, starts, and monitors the full pipeline via Docker Compose. It prints live progress (Events ingested: N) every 5 seconds and exits when ingestion is complete.

Prerequisites: Docker and Docker Compose installed. No other dependencies needed.

Configuration: Copy .env.example to .env and set your API_KEY. All other defaults work out of the box.

---

Architecture

flowchart TB
  subgraph external [External]
    API["DataSync API<br/>(paginated events)"]
  end

  subgraph producer [Producer - 1 instance]
    ProdService["producer.service.ts<br/>fetchEvents loop"]
    CursorSvc["cursor.service.ts<br/>cursor persistence"]
    ApiRepo["api.repository.ts<br/>HTTP client + rate limit parsing"]
  end

  subgraph redis [Redis]
    Stream["Redis Stream<br/>key: events:ingestion"]
    CursorKey["Key: ingestion:cursor"]
    ProgressKey["Key: ingestion:progress"]
    CompleteKey["Key: ingestion:complete"]
  end

  subgraph workers [Workers - 2 replicas x concurrent loops]
    WorkerSvc["worker.service.ts<br/>workerLoop()"]
    StreamSvc["stream.service.ts<br/>consume + ack"]
    RateLimiter["rate-limiter.service.ts<br/>token bucket"]
  end

  subgraph db [PostgreSQL]
    EventsTable["events table<br/>(UUID PK, JSONB props)"]
  end

  API -->|"cursor pagination"| ApiRepo
  ApiRepo --> ProdService
  ProdService -->|"XADD batches"| Stream
  ProdService -->|"save cursor"| CursorKey
  ProdService -->|"track count"| ProgressKey
  ProdService -->|"mark done"| CompleteKey

  Stream -->|"XREADGROUP"| StreamSvc
  StreamSvc --> WorkerSvc
  WorkerSvc -->|"XACK + XDEL"| Stream
  WorkerSvc -->|"acquire token"| RateLimiter
  RateLimiter -->|"bulk INSERT"| EventsTable
  WorkerSvc -->|"check"| CompleteKey

Loading

Docker Services

Service	Description
redis	Redis 7 Alpine -- message buffer (Redis Streams) and cursor/progress store
postgres	PostgreSQL 16 Alpine -- final event storage
producer	Single instance -- fetches events from the API and publishes batches to the stream
worker	2 replicas -- each runs concurrent loops consuming batches and inserting into Postgres

Data Flow

1. Producer fetches pages of events from the DataSync API using cursor-based pagination
2. Events are chunked into batches and published to a Redis Stream (events:ingestion) via XADD
3. The cursor and progress count are persisted in Redis for resumability
4. Workers consume one message at a time from the stream using XREADGROUP with a consumer group, which automatically load-balances messages across all worker loops
5. Each worker immediately acknowledges the message (XACK + XDEL), acquires a rate-limiter token, and bulk-inserts the event batch into PostgreSQL with ON CONFLICT DO NOTHING for idempotency
6. When the producer finishes all pages, it sets an ingestion:complete flag in Redis. Workers detect this, drain remaining messages, and exit gracefully

Error Handling

• Rate limit (429): exponential backoff up to 15 retries, respects retry-after header
• Cursor expiration (400): clears saved cursor and restarts from the beginning
• General API errors: exponential backoff up to 5 retries
• DB insert failures: batch is dropped (duplicates are safe due to ON CONFLICT DO NOTHING)
• NOGROUP errors: consumer group is automatically recreated
• Graceful shutdown: SIGTERM/SIGINT drains in-flight work before closing connections

---

API Discoveries

• Rate limit behavior is communicated via response headers: the API returns x-ratelimit-remaining, x-ratelimit-reset, and retry-after headers. The producer proactively sleeps when remaining is low to avoid hitting 429s, and uses retry-after for precise backoff when rate-limited.
• Cursors have a TTL: pagination cursors expire after some inactivity. The producer handles this by catching CURSOR_EXPIRED errors, clearing the saved cursor, and restarting from the beginning.
• Timestamp formats vary: some events have string timestamps, others have numeric. All timestamps are normalized to milliseconds (BIGINT) before insertion.
• Data contains null bytes: some string fields include \0 characters that break PostgreSQL inserts. These are stripped during the bulk insert transformation.

---

What I Would Improve With More Time

Monitoring and observability: Add Prometheus metrics (events/sec throughput, queue depth, insert latency, error rates) with a Grafana dashboard for real-time visibility into the pipeline health. Currently monitoring is limited to console logs.

Other potential improvements:

• Retry failed batches via a dead-letter stream instead of dropping them
• Use PostgreSQL COPY for higher write throughput on large batches
• Add unit and integration tests for the producer loop, worker loop, and rate limiter
• Expose a simple health-check HTTP endpoint from workers

---

AI Tools Used

Cursor was used as the AI-assisted development environment throughout this project -- for code generation, architecture decisions, debugging, and documentation.

---

Project Structure

packages/
  producer.ts                    # Producer entry point
  worker.ts                      # Worker entry point
  services/
    producer.service.ts          # API fetch loop, cursor management, batch publishing
    worker.service.ts            # Concurrent worker loops, shutdown handling
    stream.service.ts            # Redis Stream publish/consume/ack operations
    rate-limiter.service.ts      # Token bucket rate limiter for DB writes
    cursor.service.ts            # Cursor persistence and completion tracking
  repository/
    api.repository.ts            # DataSync API HTTP client, rate limit header parsing
    redis.repository.ts          # ioredis wrapper for Streams and key-value ops
    database.repository.ts       # PostgreSQL bulk insert with connection pooling
docker-compose.yml               # Redis, PostgreSQL, Producer, Worker services
init.sql                         # PostgreSQL schema (events table + indexes)
Dockerfile                       # Node.js 20 Alpine build
run-ingestion.sh                 # One-command execution and progress monitoring