A layered ETL platform turning simulated task-management activity into analytics-ready insights.
This is a real, working codebase, not a mockup — the event generator, three Postgres databases, and Python ETL pipelines all run today via Docker Compose. The analytics-serving API, dashboard, and orchestration are the pieces still ahead. Source linked below.
Business Problem
TaskPulse is built around a question every engineering manager eventually asks: which users are productive, which workspaces are falling behind, and where are tasks piling up? Answering that by querying a live operational database directly is a bad idea — it competes with production traffic and invites ad-hoc, unrepeatable analysis. TaskPulse solves it by generating realistic workspace activity (tasks, assignments, status changes) and running it through a layered pipeline — raw capture, cleaning, and analytics — so the numbers a dashboard would show come from a purpose-built, incrementally-updated store instead of a query bolted onto the live app.
Architecture
TaskPulse runs as a pnpm monorepo backed by three separate PostgreSQL 16 databases — raw, clean, and analytics — each in its own Docker container with its own Drizzle ORM schema and migration history. Because they're physically separate databases, not just separate schemas, there are no foreign keys across layers: each layer only owns what it's responsible for. A Node.js + Express service generates simulated workspace and task events and writes them into the raw database. Python workers built with pandas and psycopg2 pull from raw, validate and normalize the data, and load it into the clean database; a second set of workers aggregates clean data into analytics tables — user productivity, workspace efficiency, task completion, and project performance metrics.
Pipeline
Each pipeline stage — extractor, transformer, loader — lives in its own Python module, wired together by a pipeline entrypoint. Extraction is incremental by default: the raw-to-clean job reads a job_history table for the last successful run and only pulls rows created after that timestamp, then records its own run in the same table once it finishes. Pandas handles the transformation logic — grouping, aggregation, completion-rate math — before loading into typed, Drizzle-managed tables. The full stack — three Postgres instances, the generator, and the workers — comes up with a single docker compose up.
Architecture
9 stages, each with a single responsibility — from the event generator to a dashboard that's still on the roadmap.
Event Generator
Node.js + Express service producing simulated workspace, task, and status-change events.
Raw Database
postgres_raw — every event landed as JSON, untouched, with full replay history.
Incremental Extractor
Python job reads a job_history table and pulls only events created since the last successful run.
Cleaning & Validation
Pandas normalizes statuses, timestamps, and event types before anything moves downstream.
Clean Database
postgres_clean — typed, validated tables: workspaces, projects, tasks, task_assignees.
Aggregation Workers
Python + pandas compute productivity, efficiency, and completion metrics per workspace and user.
Analytics Database
postgres_analytics — precomputed, read-optimized metric tables.
Analytics API
Express endpoints to serve precomputed metrics to a dashboard — designed, not yet built.
Dashboard
Visualizing metrics and trends for end users — planned next.
Database Schema
Real tables from the raw, clean, and analytics databases. Relationships stay inside each database — nothing is a foreign key across a layer boundary.
Raw Database
raw_events
every incoming event, unprocessed and immutable.
job_history
tracks the last successful run per pipeline job.
Clean Database
workspaces
validated, normalized operational data.
tasks
individual units of work within a project.
task_assignees
many-to-many link between tasks and users.
Analytics Database
user_productivity_metrics
precomputed, read-optimized metrics per user.
Project Layout
Each pipeline stage gets its own module — nothing shares responsibility.
taskpulse-data-platform/
├── apps/
│ ├── api/ # Node.js + Express — event generator service
│ └── analytics-python/ # Python ETL
│ ├── extractors/ # Pull rows from raw / clean via SQL
│ ├── transformers/ # pandas cleaning + aggregation logic
│ ├── loaders/ # Write into clean / analytics tables
│ ├── pipelines/ # Entrypoints wiring extract → transform → load
│ └── worker/ # Long-running pipeline runners
├── databases/
│ ├── raw-db/ # Drizzle schema + migrations — raw_events, job_history
│ ├── clean-db/ # Drizzle schema + migrations — workspaces, tasks, users…
│ └── analytics-db/ # Drizzle schema + migrations — productivity, efficiency metrics
├── scripts/ # dev-all.ts, reset_clean_db.ts
├── docker-compose.yml # postgres_raw, postgres_clean, postgres_analytics
├── pnpm-workspace.yaml
└── package.jsonFavorite Query
The analytics database stores precomputed productivity metrics per user, per workspace — this is the kind of ranking query that table is built to support, run directly against real column names from the live schema.
SELECT
user_id,
workspace_id,
completed_tasks,
total_assigned_tasks,
completion_rate,
RANK() OVER (
PARTITION BY workspace_id
ORDER BY completion_rate DESC
) AS rank_in_workspace
FROM user_productivity_metrics
WHERE total_assigned_tasks >= 5
ORDER BY workspace_id, rank_in_workspace;Pipeline Code
Unedited functions from the actual repository — not illustrative examples.
Incremental extraction
Only pulls rows created after the last successful run — real code, not paraphrased — the incremental-load pattern the whole pipeline is built around.
def extract_raw_task_events():
conn = get_raw_connection()
cur = conn.cursor()
cur.execute("""
SELECT *
FROM raw_events re
WHERE re.entity_type = 'task' AND re.created_at >
(SELECT h.last_run_at FROM job_history h)
""")
rows = cur.fetchall()
cur.close()
conn.close()
return rowsSafe writes with job tracking
Updates job_history inside a try/finally so the connection always closes even if the write fails, and records the run so the next incremental pull knows where to start.
def load_latest_job_history(counts, started_at, finished_at, duration, message, next_run):
conn = get_raw_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE job_history
SET
last_run_at = NOW(),
started_at = %s,
finished_at = %s,
duration_ms = %s,
message = %s,
payload = %s,
next_run_at = %s
WHERE job_name = %s
""",
(started_at, finished_at, duration, message, Json(counts), next_run, "raw_to_clean_worker"),
)
conn.commit()
finally:
conn.close()Design Mockup
Illustrative design mockup — the analytics tables behind these numbers are real; this dashboard UI hasn't been built yet.
Workspace Productivity
Project Progress
User Activity
Task Completion Rate
Weekly Throughput
Design Mockup
Illustrative CLI output — shows the intended run sequence, not a captured log.
$ pnpm run pipeline:raw:clean
[RAW_TO_CLEAN] : {'PROCESSED': 128, 'TASK_CREATED': 54, 'TASK_STATUS_CHANGED': 74}
$ pnpm run pipeline:users:productivity
[USER_PRODUCTIVITY_METRICS] : [{'user_id': '01HZK...', 'completion_rate': 82.35}]
✓ job_history updated — next run scheduled
Engineering Decisions
Every layer exists for a reason — here's the reasoning behind the four that matter most.
Three physical databases, not one warehouse with schemas
Raw, clean, and analytics have different write patterns and consumers. Separate Postgres instances mean a slow analytics query can never contend with the write path capturing new events, and it's structurally impossible to blur the layers.
Incremental extraction via a job_history table
Each run reads the last recorded run time and only pulls newer rows, rather than diffing against 'now'. Reprocessing the full event log on every run doesn't scale once the raw table grows.
Drizzle ORM with a migration history per database
Each of the three databases owns its own schema and migrations, so a change to analytics tables can ship without touching raw or clean.
SQL for extraction, pandas for transformation
Joins and filtering happen in SQL where the database can use indexes; grouping, aggregation, and completion-rate math happen in pandas, where it's easier to test and iterate on.
I'm happy to walk through any part of this design in more depth — the incremental load strategy, the schema, or what I'd change next.