DEV Community: 〽️ 𝙍𝙤𝙨𝙝𝙖𝙣

How I Built a Production Observability Stack — And Broke It Twice Before It Worked

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Sun, 05 Apr 2026 13:36:40 +0000

I used to dismiss monitoring as something you bolt on after the real engineering is done. Logs were noise. Metrics were "a later problem." Alerts were for teams with dedicated SREs, not a small startup running three service types on Render.

I was wrong. Badly wrong. And it took a self-inflicted incident — where my own monitoring system became the thing that needed monitoring — to understand why observability is engineering, not afterthought.

This is a detailed account of building our observability stack from scratch: what we built, what broke, why it broke, and what the architecture looks like today.

The starting point

We run three service types on Render:

A web service — the main API and frontend server
Background workers — async job processors (queuing, retries, scheduled tasks)
Key-value stores — Render's managed Redis-compatible service

Before this project, each service logged to Render's default log drain. If something broke in production, you'd open the Render dashboard, pick a service, and scroll. No correlation across services, no metrics, no alerting, no history beyond what Render retained.

The goal was simple to state: centralize everything into a self-hosted Grafana stack, set up alerting, and gain actual visibility into what the system was doing.

The stack

Before getting into what went wrong, here's what we ended up with:

Component	Role
Caddy	Reverse proxy and automatic TLS. All external traffic enters here.
syslog-proxy	Custom Python container. Enforces token validation on inbound syslog, strips the token, forwards clean RFC 5424 to Alloy.
Vector	Disk-backed telemetry buffer between syslog-proxy and Alloy. Prevents message loss during restarts or slowdowns.
Alloy	Grafana's telemetry collector. Receives logs, applies filters and transformations, routes to the right backend.
Loki	Log storage. Stores log lines compressed and indexed by labels. Queried with LogQL.
Mimir	Metrics storage. Long-term Prometheus-compatible time-series backend.
Tempo	Trace storage. Stores distributed traces with minimal indexing — cheap to run.
Grafana	The UI. Dashboards, alert rules, and unified querying across Loki, Mimir, and Tempo.

The full data flow:

Render services
    → syslog drain (TCP, RFC 5424)
    → syslog-proxy (token validation + strip)
    → Vector (disk-backed buffer)
    → Alloy (filter, transform, route)
    → Loki (logs) / Mimir (metrics) / Tempo (traces)
    → Grafana (dashboards + alerts)
    → Datadog + BetterStack (fan-out for external alerting)

All of this runs on a single 4 GB DigitalOcean droplet, currently at ~30% total memory usage. That number matters — it means the stack is right-sized, not just "scaled until it stopped crashing."

Problem 1: No token enforcement on the log stream

Render lets you configure a syslog log drain — a TCP endpoint that receives log lines in RFC 5424 format. You give Render a URL with a token embedded in it, and Render forwards logs there.

The problem: that token is in the URL, but there's no native mechanism to validate it before your collector ingests the message. Render's outbound IP range is also shared across tenants. Any service within that range — from any organisation — can technically send data to your syslog endpoint if they know the address. And since we were pointing directly at Alloy, anything that reached the port got ingested.

This is less of an active security threat and more of a correctness and isolation problem. You want to know that the logs you're querying are your logs.

The fix: we wrote a lightweight Python container — syslog-proxy — that sits between Render and Alloy. Every inbound TCP connection goes through it first. It reads the syslog message, extracts the token from the structured data field (RFC 5424 puts auth metadata in the [SD-ID key="value"] block), validates it against a shared secret, strips it from the message, and forwards clean RFC 5424 to Alloy's syslog listener.

If the token is missing or wrong, the connection is dropped. No entry.

The key distinction: the proxy isn't adding authentication to the stream. Render was already sending a token. The proxy is the enforcement layer that was missing — the thing that actually checks it before data moves downstream.

Problem 2: Connection drops under load

After deploying the proxy, we started seeing intermittent message loss. Not every time, not obviously, but present — we'd notice gaps in log sequences that shouldn't have gaps.

The root cause was straightforward: the proxy was synchronous and had no internal buffer. When Alloy was slow to accept a connection (startup, GC pause, momentary backpressure), the proxy would drop the TCP connection rather than queue the message. Lost message, no retry, no error surfaced to the user.

The fix: we inserted Vector between syslog-proxy and Alloy.

Vector is a high-performance, Rust-based telemetry agent built specifically for this kind of pipeline work. The relevant feature here is disk-backed buffering. Messages are written to disk immediately on receipt, then forwarded to the downstream destination. If Alloy is slow, Vector queues. If Alloy restarts entirely, Vector holds the messages and delivers once the connection is re-established.

The pipeline now looks like: syslog-proxy validates and strips → hands off to Vector → Vector buffers to disk → Vector forwards to Alloy.

Connection drops stopped immediately. We have not lost a log message to transport failure since.

The incident

With the pipeline stable, we validated it end-to-end on our highest-volume service — the web API. It processed tens of thousands of log lines without issue. Alloy filtered and routed correctly. Loki ingested cleanly. Grafana showed the data. Everything looked good.

So we did what seemed logical: we added all remaining services to the syslog drain simultaneously.

Within two hours, the system was on fire.

Both Datadog and BetterStack — external services we fan logs to in parallel for alerting and long-term retention — were overwhelmed. The Datadog exporter started returning unexpected EOF from their intake API. This is Datadog actively closing connections, not just timing out. They were rejecting us.

Alert rules fired across the board. The alert channel, which was supposed to tell us when our application was unhealthy, was now full of noise about the monitoring pipeline being unhealthy. The observer had become the thing being observed.

We rolled back all drain configurations to Render's defaults within minutes.

Post-mortem

The root cause was a combination of three things that compounded each other:

1. Unfiltered background worker logs. Background job processors emit a lot of lifecycle noise by default — job enqueued, job started, job completed, job retried, and so on. These events fire on every job. At our job volume, a single background worker service can generate thousands of log lines per minute. None of this is signal for ongoing observability. It's useful when debugging a specific job failure, not for dashboards.

2. Fan-out multiplication. We were routing logs to three destinations simultaneously: Loki, Datadog, and BetterStack. A 3x spike in log volume becomes a 9x spike in outbound pipeline throughput. Every destination gets hit at the same time, with the same burst.

3. All services at once. Testing on one service — even the highest-volume one — told us the pipeline could handle a single source. It told us nothing about what happens when five sources open simultaneously. The aggregate volume was an order of magnitude higher than anything we'd validated.

What we changed

Incremental rollout. After rolling back, we added services back one at a time with 24 hours of observation between each. If volume, error rates, and downstream health looked stable for 24 hours, we added the next one. The full rollout took five days instead of five minutes.

Log filtering in Alloy. We defined explicit filter rules per service type. Background worker INFO-level lifecycle events (enqueued, started, completed) are now dropped by Alloy before they reach any destination. Only WARNING and above, plus specific job failure patterns, pass through. This cut background worker log volume by roughly 80%.

Volume alerts on the ingestion pipeline. We added alert rules on Alloy's own metrics — ingestion rate, error rate, downstream write failures. If the pipeline itself starts showing stress, we know before it cascades.

The detail that almost cost us — the 8 KB syslog limit

After the system was stable, we noticed something odd: certain application errors were present in Render's own log view but missing from Loki. Not all of them. Just the large ones.

The ones that were missing were stack traces. Full Java-style stack traces can easily exceed 100 KB. Syslog has a default maximum message size of 8 KB (defined in RFC 5424). Messages that exceed it are silently truncated or dropped, depending on the implementation.

Silent. No error. No warning. No indication in any metric that a message was lost. The data simply didn't arrive.

We increased the max_message_len parameter in the syslog-proxy and Alloy's syslog listener to 256 KB. The missing stack traces appeared immediately.

The lesson here is broader than syslog: check the defaults of every tool in your pipeline. Buffer sizes, message limits, timeout values, retry caps — these are all set to something that was reasonable for a generic use case. They may not be reasonable for yours. And when they're wrong, most tools will not tell you.

Current state

The stack has been running stably for several weeks. Current metrics:

Droplet: 4 GB RAM, ~30% utilization under normal load
Log ingestion: all three service types, filtered, continuous
Storage: local filesystem (S3 planned for long-term retention)
External fan-out: Datadog and BetterStack, with volume alerts before we approach their intake limits
Alert coverage: application error rates, pipeline health, job failure patterns, infrastructure metrics via Mimir

The Grafana dashboards now show a real-time view of system health across all services. When something breaks, we know about it from an alert before a user reports it. That has happened twice since launch, and both times the alert fired before any user-facing degradation was detectable.

Key takeaways

Not all logs are worth storing. Define what matters per service before you start ingesting. Background worker lifecycle events are not observability — they're debug information that belongs in a trace, not a log aggregator. Decide at collection time, not retention time.

Buffer every I/O boundary. Silent drops are worse than backpressure. Vector saved us from losing data during restarts and slowdowns. Put a buffer anywhere data moves between two systems that can fail independently.

Fan-out multiplies everything. A 3x spike in log volume doesn't hit three destinations at 3x each — it hits three destinations at 3x simultaneously, and each downstream system now has to handle the same burst. Design for the aggregate.

Roll out incrementally. One source, observe for 24 hours, then next. Validating on a single service tells you almost nothing about aggregate behavior. The incident would not have happened if we had added services one at a time.

Check your defaults. The 8 KB syslog limit is a good example of a default that works for most cases and silently breaks for edge cases. Every tool in your pipeline has limits like this. Read the config reference. Set explicit values.

Monitor the monitor. Your observability pipeline is a production system. It needs its own health metrics, its own alerts, and its own runbook. If the pipeline goes down during an incident, you're blind exactly when you need visibility most.

Stack

Grafana · Loki · Mimir · Tempo · Alloy · Vector · Caddy · Render · DigitalOcean · S3 (planned)

If you're building something similar or have questions about any part of the architecture, feel free to reach out.

From Backend Engineer to Building Production Infrastructure in AWS

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Sun, 05 Apr 2026 13:30:01 +0000

Last year in September, I was laid off. By November, I'd landed a role at a startup with a responsibility I wasn't fully prepared for: build a reliable AWS infrastructure from scratch.
I'd only ever worked as a fullstack engineer. I knew the basics — VPCs, load balancers, containers — but not at the level where I'd feel confident deploying anything to production, let alone architecting an entire infrastructure with a real budget on the line.
But I took the job anyway. This is the story of what I did, what I got wrong, and what I learned along the way.

This is the story of every mistake, pivot, and small win along the way.

The Starting Point: Rewriting Before Deploying

Before I could deploy anything, there was a bigger problem. The existing backend was legacy JavaScript — no types, no structure, no clear separation of concerns. Deploying it as-is would just be putting broken code on expensive servers.

So I rewrote it in TypeScript.

I went with a 3-layer architecture: routes, controllers, and module functions. The goal was maintainability — whoever touches this codebase next should be able to find things without a treasure map. Most of the refactoring happened through Cursor, which made the migration significantly faster than doing it by hand.

With the codebase stable, it was time to ship.

First Deployment: The ECS Experiment

The first service that needed to go live was a CMS — a monorepo running a backend with SQLite. My initial plan was to standardise on ECS for everything. It felt like the "right" way to do containers on AWS.

I containerised the backend, learned how ECR works, pushed the image, mounted an EBS volume for the SQLite database, and put an ELB in front of it. It took two full weekends. But it worked.

Then I got the message from management: the cost of ECS is too high.

For a simple CMS serving internal traffic, they were right. I was over-engineering it.

The Pivot: Right-Sizing the Infrastructure

I tore everything down and went back to research — blogs, documentation, conversations with LLMs to pressure-test ideas. The answer was simple once I stopped thinking in terms of "what's the most modern approach" and started thinking about "what does this workload actually need."

A CMS doesn't need container orchestration. It needs a box that runs.

Here's what I landed on:

Backend: EC2 t3.small — cheap, sufficient, easy to manage
Frontend: AWS Amplify — fast deploys, built-in CDN, no server to maintain
Database: SQLite on EBS with weekly backups to S3

Total cost: a fraction of the ECS setup. Same reliability for the workload. That was my first real win — not building the most impressive architecture, but building the right one.

The Docker Build Problem

With the infrastructure sorted, a new bottleneck appeared: CI/CD.

Every push triggered a Docker build, and each build was taking around 5 minutes. We were running a monorepo with pnpm, and the issue was layer caching — or rather, the lack of it. Every build was reinstalling every dependency from scratch.

The fix was multi-stage builds. By separating the dependency installation layer from the application code layer, Docker could cache the expensive pnpm install step and only rebuild what actually changed.

Build times dropped from 5 minutes to 20 seconds. Across 4 repositories. That's the kind of improvement that compounds — every developer, every push, every day.

Nightly API Testing Without a Dedicated Server

The next requirement was daily test runs — a nightly job that would hit all our APIs, run the test suite with Jest and Supertest, and generate a report for the team.

My first instinct was to spin up another EC2 instance. But that meant paying for a machine that sits idle 23 hours a day, plus managing its uptime, patching, and monitoring.

Then I discovered that GitHub Actions provides temporary VMs for exactly this kind of job. No infrastructure to manage. No idle costs. The workflow spins up, runs 100+ API tests in about 5 minutes, generates an HTML report, uploads it to S3, and notifies the admins.

A scheduled cron job in a YAML file replaced an entire server.

What I Actually Learned

Looking back at these four months, the technical skills were the easy part. Docker, ECS, EC2, Amplify, GitHub Actions — you can learn any of these in a weekend. The harder lessons were about decision-making:

Cost awareness changes how you architect. When you're spending someone else's money, every decision carries weight. "Best practice" doesn't mean anything if it's 10x the cost for a workload that doesn't need it.

The first solution is rarely the right one. ECS was a fine technology. It was the wrong choice for the problem. Being willing to tear something down after two weekends of work is a skill in itself.

Boring infrastructure is good infrastructure. An EC2 instance with a cron job isn't exciting. It's also not going to wake you up at 3am.

Optimise the inner loop. The Docker build improvement saved 4+ minutes per push across 4 repos. Over a team and a quarter, that's days of developer time recovered.

I'm not a DevOps engineer. I'm not a backend engineer either. Honestly, it's better to let go of that mindset entirely. In the coming times, just doing one thing won't be enough. You're an engineer — you solve problems. The label doesn't matter; the willingness to figure it out does.

Months in and still going, I've learned that the gap between
"knowing the theory" and "running it in production" is mostly filled with wrong first attempts and the willingness to start over.

If you're in a similar position — thrust into infrastructure work without a roadmap — my advice is simple: deploy something, get it wrong, and iterate. The cloud makes it cheap to experiment, and every teardown teaches you more than any tutorial.

# Schema Evolution & Encoding: Building Future-Proof Data Systems 🚀

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Fri, 12 Sep 2025 11:35:41 +0000

How to design data-intensive applications that gracefully handle schema change

The Challenge: Data That Evolves 📈

In modern distributed systems, schema changes are inevitable. Your application will grow, requirements will shift, and your data structures must adapt. But here's the catch: you need to maintain compatibility with existing data and older versions of your application.

Understanding Compatibility Types 🔄

Backward Compatibility ⬅️

New code can read data written by old code
Essential when upgrading applications gradually

Forward Compatibility ➡️

Old code can read data written by new code
Critical for rolling back deployments safely

Encoding = Serialization 📦

Encoding is the process of converting in-memory data structures into a byte sequence that can be:

Stored in files
Transmitted over networks
Processed by different systems

Think of it as packaging your data for shipping! 📮

Traditional RPC Solutions 🌐

gRPC & Apache Thrift Approach

// Protocol Buffers example
message UserProfile {
  int32 user_id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4 [optional]; // New field - backward compatible!
}

Key Strategies:

✅ Add new fields as optional
✅ Assign default/null values to new fields
✅ Generate classes from schema definitions
✅ Perfect for statically typed languages (Java, C++, Go)

The Dynamic Language Problem 🐍

Challenge: Traditional schema-based systems don't play well with dynamic languages like JavaScript, Python, or Ruby.

Why?

No compile-time type checking
Runtime schema validation needed
Class generation feels unnatural

Apache Avro 🌟

Apache Avro solves the dynamic language compatibility problem with a genius approach: dual schemas!

Avro's Dual Schema System 🎭

Writer Schema 📝

The schema used when encoding the data

Reader Schema 📖

The schema used when decoding the data

// Writer Schema (v1)
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"}
  ]
}

// Reader Schema (v2) - with new field
{
  "type": "record", 
  "name": "User",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}

How Avro Handles Compatibility ⚡

Field Resolution Rules:

Field in writer but not reader → Ignored 🙈
Field in reader but not writer → Default value used 📋
Field in both → Direct mapping ✅

This elegant system enables both forward and backward compatibility!

Schema Negotiation in Practice 🤝

When two processes communicate over a bidirectional network connection:

sequenceDiagram
    participant A as Service A
    participant B as Service B

    A->>B: Connection Request + Schema Version
    B->>A: Schema Negotiation Response
    A->>B: Agreed Schema for Session
    Note over A,B: Use negotiated schema for<br/>entire connection lifetime

Benefits:

✅ Both sides agree on schema version upfront
✅ Optimal performance (no per-message overhead)
✅ Clear compatibility guarantees

Best Practices for Schema Evolution 📚

Do's ✅

Always add new fields as optional
Provide sensible default values
Use semantic versioning for schemas
Test compatibility with real data
Document migration strategies

Don'ts ❌

Never remove required fields
Avoid changing field types drastically
Don't reuse field IDs/names for different purposes
Never skip compatibility testing

Real-World Impact 🌍

Companies using these patterns:

Netflix: Schema evolution for microservices
LinkedIn: Avro for data pipelines
Uber: Protocol Buffers for service communication
Airbnb: Thrift for cross-service APIs

The result? Zero-downtime deployments and seamless data migrations! 🎉

Conclusion: Future-Proof Your Data 🔮

Schema evolution isn't just a technical detail—it's a business enabler. By choosing the right encoding strategy:

Reduce deployment risks ⚡
Enable continuous delivery 🚀
Support diverse technology stacks 🌈
Maintain system reliability 🛡️

Key Takeaway: Whether you choose Protocol Buffers, Thrift, or Avro, the principles remain the same—design for change from day one!

Validating Environment Variables in Node.js with Zod

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Thu, 22 May 2025 20:00:01 +0000

Ever pushed a project to production only to realize an environment variable was missing or mistyped? I have—and it wasn’t fun.

Recently, I ran into a subtle but recurring issue in one of our Node.js projects: environment variables not loading correctly from the .env file. At first glance, this seemed trivial—just use dotenv, right? But as our configuration grew more complex, undefined values started creeping in silently, breaking features without clear errors.

The Problem with Plain .env + process.env

Like many teams, we began with this simple setup:

import dotenv from "dotenv";
dotenv.config();

const CORE_DB_URI = process.env.CORE_DB_URI;

But the issues quickly stacked up:
• We forgot to set some variables in staging or dev environments.
• Typos in variable names would go unnoticed until runtime.
• Some variables like PORT needed to be parsed (parseInt), while others had fallbacks—lots of boilerplate.

Worst of all, we had no way to know what was missing or invalid until something failed. That’s not a great developer experience.

The Breakthrough: Validating the Environment with Zod

After some research and trial-and-error, we discovered a better way—define a schema for environment variables using zod, and parse process.env against it before anything else runs.

Here’s what that looks like:

import z from "zod";

const envSchema = z.object({
  NODE_ENV: z.string().optional(),
  HOST: z.string().optional(),
  PORT: z.string().transform(val => parseInt(val)).optional(),

  MONGO_DB_URI: z.string().nonempty("MONGO_DB_URI is required"),
  // ... add other required vars

  AWS_ACCESS_KEY: z.string().nonempty("AWS_ACCESS_KEY is required"),
  SMTP_USERNAME: z.string().nonempty("SMTP_USERNAME is required"),
});

Then we updated the config loader like this:

import dotenv from "dotenv";
import { envSchema } from "./envSchema";

dotenv.config();

let parsedEnv;
try {
  parsedEnv = envSchema.parse(process.env);
  console.log("✅ Environment validation passed");
} catch (err) {
  console.error("❌ Environment validation failed:");
  console.error(JSON.stringify(err.format?.(), null, 2));
  process.exit(1);
}

Benefits We Gained

This refactor gave us immediate wins:
• Validation at startup: The app won’t even run unless all required env vars are present.
• Helpful errors: Each missing variable now shows a clear error message.
• No more implicit casting or null-checks: Everything is validated and transformed up front.
• Cleaner config code: No more process.env.XYZ || "fallback" everywhere.

Instead of this:
const CORE_DB_URI = process.env.CORE_DB_URI!;

We now use this:
const config = { CORE_DB_URI: parsedEnv.CORE_DB_URI, };

Fail Fast, Not in Production

One thing that really sold me on this pattern
[https://www.youtube.com/watch?v=eZBsBMf5zuQ]
is how it catches config issues at runtime—before your app misbehaves in production.

Without schema validation, a missing environment variable might not throw an error until it’s used—say, when connecting to a database or sending an email. That’s late. By then, users might already be affected.

With zod, your app won’t even boot if a required environment variable is missing or misconfigured. That’s intentional. You get a clear error in development, staging, or CI pipelines—not after you’ve shipped broken code.

It’s simple, safe, and clean.

Final Thoughts

It’s easy to underestimate the importance of config hygiene—until it breaks something. Using zod for schema validation gives you type safety, meaningful errors, and cleaner code. It’s now a pattern I’ll carry into every serious Node.js project.

If you’re still relying solely on .env + process.env, try wrapping it with validation. Your future self—and your team—will thank you.

From Storage to Stream: Delivering MongoDB Data Directly to Users

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Mon, 30 Dec 2024 20:16:48 +0000

Step 1: MongoDB Cursor

Here’s how we set up the cursor (reusing your snippet):

const cursor =
    userObject?.data?.serviceProviderName === 'ZYRO'
        ? zyroTransactionModel.find(query).cursor()
        : finoTransactionModel.find(query).cursor();

console.log("Cursor created successfully");

Step 2: Setting Up the ZIP File
Use the yazl library to stream CSV data into a ZIP file:

const yazl = require('yazl');
const zipfile = new yazl.ZipFile();

reply.raw.writeHead(200, {
    "Content-Type": "application/zip",
    "Content-Disposition": "attachment; filename=transactions.zip",
});

zipfile.outputStream.pipe(reply.raw);

const cleanup = async () => {
    console.log("Cleaning up resources...");
    zipfile.end(); // Finalize ZIP
    await cursor.close();
};
reply.raw.on("close", cleanup);
reply.raw.on("error", cleanup);

Step 3: Creating Dynamic CSV Streams
Generate CSV data dynamically and stream it into the ZIP file:

const createNewCSVStream = (headers) => {
    const csvStream = new Readable({ read() {} });
    csvStream.push(headers.join(",") + "\n"); // Add headers
    return csvStream;
};

const filteredHeaders = getHeaders(transactionDownloadFields, userObject?.state?.auth?.role);
const currentCSVStream = createNewCSVStream(filteredHeaders);

zipfile.addReadStream(currentCSVStream, "transactions_part_1.csv");

Step 4: Streaming MongoDB Data to CSV
Stream the data from MongoDB directly into the CSV:

cursor.on('data', (doc) => {
    const csvRow = filteredHeaders.map(header => doc[header.key] || '').join(',');
    currentCSVStream.push(csvRow + '\n'); // Write row
});

cursor.on('end', () => {
    currentCSVStream.push(null); // End the stream
    zipfile.end(); // Finalize the ZIP
});

Step 5: Processing Data from MongoDB Cursor
Stream documents from the MongoDB cursor, transform them as needed, and dynamically write rows to the CSV stream:

try {
    for await (const doc of cursor) {
        if (clientDisconnected) {
            console.log("Client disconnected. Stopping processing...");
            break;
        }

        streamedCount++;
        rowCount++;

        let row = "";
        const filteredHeaders = getHeaders(
            transactionDownloadFields,
            userObject?.state?.auth?.role
        );

        for (let i = 0; i < filteredHeaders.length; i++) {
            const field = filteredHeaders[i];

            // Fetch the corresponding field configuration from transactionDownloadFields
            const originalField = transactionDownloadFields.find((f) => f.value === field.value);

            // Get the value from the transaction document
            let value = getValueFromTransaction(doc, field.value);

            // Apply transformation if the field has a transform function
            if (originalField?.transform) {
                value = originalField.transform(value);
            }

            // Enclose the value in double quotes
            value = value !== undefined ? `"${value}"` : '"N/A"';
            row += (i > 0 ? "," : "") + value;
        }
        row += "\n";
        currentCSVStream.push(row);

        // Check if the row count has reached the threshold for the current CSV file
        if (rowCount >= MAX_ROWS_PER_FILE) {
            console.log(`Threshold reached for file ${fileIndex - 1}. Starting new file...`);
            currentCSVStream.push(null); // End the current CSV stream
            currentCSVStream = createNewCSVStream(); // Start a new stream
            rowCount = 0; // Reset the row count
        }
    }

    // Finalize the current CSV stream if it has data
    if (currentCSVStream) {
        currentCSVStream.push(null);
    }

    // Finalize the ZIP file
    zipfile.end();
    console.log(`Successfully streamed ${streamedCount} rows across ${fileIndex - 1} files.`);
} catch (error) {
    console.error("Error during processing:", error);
    if (!headersSent) reply.status(500).send({ error: "Failed to generate ZIP file" });
} finally {
    // Cleanup: Close the MongoDB cursor
    await cursor.close().catch((err) => console.error("Error closing cursor:", err));
}

Summary

Document Iteration Using for await...of:

Streams documents one by one from the MongoDB cursor efficiently.
Enables real-time processing without loading all data into memory.

Dynamic CSV Row Generation:

Constructs each row dynamically by iterating over filteredHeaders.
Applies transformations using a transform function, if defined in transactionDownloadFields.
Row Threshold and File Splitting:

Monitors the row count against the threshold (MAX_ROWS_PER_FILE).
Ends the current CSV stream and starts a new one when the threshold is reached.

Error Handling:

Logs and sends an error response if an issue occurs during processing.
Ensures proper cleanup by closing the MongoDB cursor in the finally block.

Finalizing Streams:

Pushes null to terminate the current CSV stream.
Completes the ZIP file once all rows are processed.

Handling Multiple requests with Redis and Bullmq

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Sat, 13 Apr 2024 18:21:38 +0000

In our previous post, we explored how to efficiently download a large dataset from MongoDB. We utilised the built-in fs module to create temporary files on the server, enabling us to handle substantial amounts of data. However, upon deploying our solution for testing, an unexpected issue arose during load testing.

Our tester discovered that when multiple requests were made to the download API within a short span of time, both requests received the same data, despite requesting different datasets. For instance, if User A requested data for January, and User B requested data for June, both users received data for January. This discrepancy indicated a critical flaw in our implementation: it was not processing requests sequentially; rather, it was executing them concurrently.

Instead of queuing requests and processing them one by one, our system was handling multiple requests simultaneously,
To address this issue and ensure the integrity of our data,
we need to implement a queuing mechanism that processes requests sequentially.

To implement the mentioned mechanism we used

Redis
Bullmq

First of all we need to set up a Redis server for that you install it locally or you can use a docker image and run it
locally

version: "3"
services:
redis:
image: redis:6.0.7
container_name: redis
restart: always
volumes:
- redis_volume_data:/data
ports:
- 6379:6379
redis_insight:
image: redislabs/redisinsight:latest
container_name: redis_insight
restart: always
ports:
- 8001:8001
volumes:
- redis_insight_volume_data:/db
volumes:
redis_volume_data:
redis_insight_volume_data:

After the server is running is successfully we will install
Bullmq

npm install bullmq

BullMQ is a lightweight, robust, and fast NodeJS library for creating background jobs and sending messages using queues. BullMQ is designed to be easy to use, but also powerful and highly configurable. It is backed by Redis, which makes it easy to scale horizontally and process jobs across multiple servers

By default it has a port 6379, so we first we need to make connection to this running server and it's simple with bullmq

const { Queue } = require("bullmq");

const connectionOpts = {
  port: 6379,
  host: "localhost", // for production it will be different
};
const downloadQueue = new Queue("download-example-queue", {
  connection: connectionOpts,
});

you can define a queue name and this will create a queue for you, so we are using Redis as a message broker to handle job queues.

As i said before we can do many things with Bullmq you learn about some methods through this blog but we will most be doing two things in our implementation, Creating Jobs and Processing Jobs,

Creating Jobs: You can create jobs by pushing them into a queue using the add method. These jobs are stored in Redis until they can be processed.

Processing Jobs: Workers pull jobs from the queue and process them. The process method specifies what to do with each job.

const addToQueue = async (exampleData) => {
  try {
    const job = await downloadQueue.add("download-example-queue", reportData, {
      removeOnComplete: true,
      removeOnFail: true,
    });
    console.log("SUCCESS ADDED TO ID", job.id);
    return job.id;
  } catch (error) {
    console.error("Error adding job to queue:", error);
    throw error;
  }
};

So here we have created a function that will take in exampleData and add that to our queue
after completion it will will removed

removeOnComplete: true, removeOnFail: true,

Additionally, we have created two functions to pause the queue and resume the queue

const pauseQueue = async () => {
  downloadQueue.pause();
};

const resumeQueue = async () => {
  downloadQueue.resume();
};

module.exports = {
  addToQueue,
  pauseQueue,
  resumeQueue,
};

We will be using the above functions in our main endpoint that we are hitting, which the place we are sending our request headers from the client

module.exports.downloadTransactionsServer = async (req, res) => {
  async function queueFunction() {
    try {
      // Extract necessary request headers
      const requestData = {
        dateRange: req.headers.daterange,
        client_id: req.headers.client_id,
        created_by: req.headers.created_by,

      };

      // Add request data to the queue and get the job ID
      const jobId = await addToQueue(requestData);

      // Check if the response status code is 200
      if (res.statusCode === 200) {
        // Return the job ID
        return jobId;
      }
    } catch (error) {
      console.log(error, "ERROR");
    }
  }

  // Call the queue function
  await queueFunction();

  // Return success response
  return createSuccessResponse(
    STATUS_CODES.ok,
    "Download initiated"
  );
};

import those addToQueue functions to your current file,
but we need to have a Worker,

Workers are the actual instances that perform some job based on the jobs that are added in the queue. A worker is equivalent to a "message" receiver in a traditional message queue. The worker's duty is to complete the job. If it succeeds, the job will be moved to the "completed" status. If the worker throws an exception during its processing, the job will automatically be moved to the "failed" status.

In microservices architecture, it's best practice to have one worker per service to ensure sequential job processing and prevent duplicates. However, scaling workers may be necessary for increased workload or fault tolerance, requiring careful management to maintain system integrity.

So, what was happening was that I had the worker created in a separate file. Due to that, multiple workers were created; however, as many requests as we made, that many workers were made. But it was working just fine; our local Redis server was able to handle it. However, in production, things were not working as expected because we had only one Redis server, and it was not creating multiple workers. Download requests would stop after coming to the queue, and they would be in a forever paused state. So, therefore, we initialised our worker in server.js.

const start = async () => {
  try {
    await fastify.listen({ port: process.env.PORT, host: "0.0.0.0" });
    const myWorker = new Worker(
      "download-example-queue",
      async (job) => {
        try {
          const reportData = JSON.stringify(job.data);
          await pauseQueue();
          await generateReportAndSendMail(JSON.parse(reportData));
        } catch (error) {
          console.error("Error processing job:", error);
        } finally {
          await resumeQueue();
        }
      },
      {
        limiter: {
          max: 1,
          duration: 1000,
        },
        ttl: 600000,
        connection: {
          port: 6379,
          host: "localhost",
        },
      }
    );
    myWorker.on("completed", (job) => {
      console.log(job.id, "HAS been completed");
    });
  } catch (e) {
    process.exit(1);
  }
};

here imported our function generateReportAndSendMail, the worker is configured to handle a maximum of one job concurrently and has a time limit of 1000 milliseconds for processing each job. It establishes a connection to a Redis server running on localhost with a TTL (time-to-live) of 600000 milliseconds. Furthermore, an event listener is implemented to log successful completion of jobs. Error handling mechanisms are incorporated to manage any potential issues during server startup or job processing. If errors occur, the process is terminated with an error code of 1 to signify the occurrence of an error.

And there you have it with that our queue is implemented and running successfully,
Thank you for reading and Follow me on LinkedIn if you’re interested in Web Development. 😊

How to zip Large Set of Data from MongoDB and upload to S3

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Sat, 06 Apr 2024 12:31:22 +0000

Starting this new year, the project we were working on received a new requirement: allowing our users to download 1 month's worth of data as a CSV file. The minimum amount of data for one month was 30 to 50 lakhs or 3 to 5 million transaction records.

Currently, our users were only able to download 1 day's worth of data, which amounted to about 30k to 40k thousands records. If they attempted to download more than that, they would encounter the same issue.

FATAL ERROR: CALL_AND_RETRY_LAST Allocation failed - JavaScript heap out of memory

At first, we thought it was simply a memory usage limit issue. So, we attempted to solve it by using below script in our package.json.

"scripts": {
    "start": "cross-env NODE_OPTIONS=--max_old_space_size=4096 webpack"
}

However, this didn't resolve the problem. Therefore, we decided to try streaming the data in our file, compressing it, and then sending it to the user.

const zlib require('zlib');

exports.downloadResource = (res, fileName, fields, data)
const json2csv = new Parser({ fields });
const csv = json2csv.parse(data);


const gzip = zlib.createGzip();

res.header("Content-Type", "text/csv");
res.header("Content-Encoding", "gzip");
=> {
res.header("Content-Disposition", `attachment; filename=${fileName}.gz`);

const buffer = [];


gzip.on('data', (chunk) => {
buffer.push(chunk);
});

gzip.on('end', () => {
for (const chunk of buffer) {
}
});
res.write(chunk);
res.end();

gzip.write(csv);

gzip.end();
};

Here, we used zlib [https://www.npmjs.com/package/zlib] to compress our files. Basically, what we tried to do was push the data to our buffer array as long as the stream of data was available. We were able to stream up to 1 million rows of CSV data. However, we encountered the same error again:

FATAL ERROR: CALL_AND_RETRY_LAST Allocation failed - JavaScript heap out of memory

This happened because storing all the data in memory at once can consume a significant amount of memory, which is what we were doing here.

If you have any suggestions, please comment below, as I couldn't figure out a solution at that time.

Since we couldn't solve the problem that way, we considered an alternative solution. It's simple: we will write the files to our server using the fs module, zip them, upload them to S3, and then send the S3 link via email to our client.

In this way we would not get any heap out issue, Here's a small flow for our solution:

Step 1:

Create a file with the fs module. Essentially, we will set up our aggregation pipeline through which our data will flow.

    const pipeline = [
      { $match: condition },
      { $sort: { modified_at: -1 } },
      {
        // Your Aggregation pipeline
      },
    ];

We will write that data into the temporary file. However, we encountered a problem: we cannot write all the data into a single CSV file. It is known that Excel sheets can display a maximum of 1 million or 10 lakh rows, so we could only add up to 1,000,000 rows of data.

Therefore, we came up with the idea to write 1 million rows in each file. If there's more data, we will write it to a new file, but each file will have a maximum of 1 million rows.

      let batchIndex = 1;
      let currentBatch = [];
      let length = 0;
      const Rows = 1000000;

      const createCsvFile = (batchIndex, headers) => {
        const csvFilePath = path.join(
          outputDirectory,
          `some_transaction_batch_${batchIndex}.csv`
        );
        return csvFilePath;
      };

      const writeCsvBatch = async (batch, index) => {
        const filePath = createCsvFile(index);
        const csvBody =
          filteredHeaders?.map((header) => header.label).join(",") +
          "\n" +
          batch.join("\n");

        await fsPromise.writeFile(filePath, csvBody);
      };

It might be hard to understand at first glance, so let me explain what's going on here.

Firstly, const outputDirectory = "Operation" + randomUUID(); generates a unique output directory name using the randomUUID() method from the Crypto interface.
This method generates a v4 UUID (Universally Unique Identifier) using a cryptographically secure random number generator.

Next, we have the createCsvFile function. This function takes an index for a batch and possibly a set of headers. It constructs a file path for a CSV file specific to that batch index within the given output directory. However, it doesn't actually create the CSV file itself; it only returns the path where it would be created.

Then, we have the writeCsvBatch(), which utilizes the createCsvFile(). It writes data into these files, with each file capable of storing up to a maximum of 1,000,000 rows. If the data exceeds this limit, it will write to a new file. This process is handled within this function.

      for await (const doc of data) {
        const data = writeCSVRow(doc, filteredHeaders);
        currentBatch.push(data);
        length++;
        if (currentBatch.length >= Size) {
          await writeCsvBatch(currentBatch, batchIndex++);
          currentBatch = [];
        }
      }

What if the data is less than 1 million of rows or just little more than a million but not 2 million or no data is there in the DB, for such cases we added some small validation

     if (length === 0) {
        return createErrorResponse(
          STATUS_CODES.forbidden,
          "No data to download"
        );
      }

      // Write the remaining records to the last CSV file
      if (currentBatch.length > 0) {
        await writeCsvBatch(currentBatch, batchIndex);
      }

Step 2:

Now that we've obtained the files and written our data to them, we need to zip them using the gzip archiver [https://www.npmjs.com/package/archiver].

The reason for this is that the files alone would be very large. For example, a single CSV file with 1 million rows had a file size of over 500 MB. This size could increase depending on the type of data being written. When we tested it with 3 million rows, the file size was almost 2 GB. Given that the server cannot handle such large files, we need to compress them. After zipping, our file size was reduced to 15 MB.

const archive = archiver("zip", { zlib: { level: -1 } });

const outputZipStream = fs.createWriteStream(outputZipPath);

archive.pipe(outputZipStream);

for (let i = 1; i <= batchIndex; i++) {
  const csvFilePath = path.join(
    outputDirectory,
    `transaction_batch_${i}.csv`
  );

  archive.append(fs.createReadStream(csvFilePath), {
    name: `transaction_batch_${i}.csv`,
  });
}

const finalizePromise = new Promise((resolve, reject) => {
  archive.finalize();
  outputZipStream.on("close", () => {
    resolve();
  });
  archive.on("error", (err) => {
    reject(err);
  });
});

await finalizePromise;

archive first we initialising this archiver object for creating a zip file.

outputZipStream This line creates a writable stream to the desired output zip file specified by outputZipPath, after creation of all the csv files and zipping them into a we will create a zip file in our server

const outputZipPath = new Date().toISOString() + ".zip";
we are using toISOString() so that every generated zip file will be unique

archive.pipe This line pipes the output of the archiver to the output zip stream, which means that data produced by the archiver will be written to the output zip file.

for loop iterates over the batch indices, reading each corresponding CSV file from the file system and appending it to the zip archive with a unique name.

const finalizePromise = new Promise((resolve, reject) => { ... })
This block wraps the finalization process of the zip archive creation in a promise. It listens for the "close" event of the output zip stream, indicating that the archiving process has finished writing to the zip file. If any errors occur during the archiving process, they are caught and rejected.

await finalizePromise: This line waits for the promise to resolve, indicating that the zip archive has been successfully created and finalised.

all these will create a zip archive containing all generated CSV files are zipped into a single archive file

step 3:

upload to s3 bucket, for this step we have a simple function that we are gonna use

const AWS = require("aws-sdk");
const fs = require("fs");

const credentials = {
  accessKey: process.env.AWS_ACCESS_KEY,
  secret: process.env.AWS_SECRET_ACCESS_KEY,
  bucketName: process.env.DOCS_BUCKET_NAME,
};

const s3 = new AWS.S3({
  accessKeyId: credentials.accessKey,
  secretAccessKey: credentials.secret,
});

const uploadFileYt = async (fileName) => {
  const fileContent = fs.readFileSync(fileName);


  const params = {
    Bucket: process.env.DOCS_BUCKET_NAME, 
    Key: fileName,
    Body: fileContent,
  };

  const data = await s3.upload(params).promise();
  return data?.Location;
};

module.exports = {
  uploadFileYt,
};

The function reads file content, constructs upload parameters, calls the S3 upload method, and returns the URL of the uploaded file.

step 4:

Send the link client, the link we get from our s3 bucket that we can send it to our client via a mail or you can directly let the user have that zip file when he clicks on download but as per my requirements we had to send it our users via mail...

const aws = require("aws-sdk");
const nodemailer = require("nodemailer");

exports.sendEmail = (
  toAddresses = ["example@example.com"],
  subject = "Example subject",
  html = `Example body`,
  ccAddresses = [],
  replyToAddress = ["example@example.com"]
) => {
   // Your logic for sending mail
  }
};

step: 5

Remove them after completion, with this i meant the folder and zip that we will create in our server, so after everything is set and done then we are finally removing them.

finally {
    fs.rm(outputDirectory, { recursive: true, force: true }, (err) => {
      if (err) {
        throw err;
      }
    });
    if (fs.existsSync(outputZipPath)) {
      // File exists, so unlink it
      await fs.promises.unlink(outputZipPath);
      console.log(`${outputZipPath} successfully removed.`);
    }
    if (client) client.close();
  }

fs.rm(outputDirectory, { recursive: true, force: true }): This line removes the outputDirectory and all its contents recursively with force, meaning it deletes files and directories forcefully without prompting for confirmation.

await fs.promises.unlink(outputZipPath): This line deletes the outputZipPath file asynchronously using promises.
if (client) client.close(): This line closes the MongoDB client if it exists.

For us we are directly taking data from our MongoDb Database's collection but for you it might be different you can do it with mongoose too but we found this solution much faster and it fulfilled our need,

incase anything fails it will delete those temporary files because finally will execute no matter what...

What if multiple requests hit at the same time, if you are interested then you can checkout this blog of mine where have implemented the solution to above problem.

Thank you for reading and Follow me on LinkedIn if you’re interested in Web Development. 😊

Open The Modal on click of browser's back button in react (with native javascript)

〽️ 𝙍𝙤𝙨𝙝𝙖𝙣 — Thu, 21 Dec 2023 18:08:11 +0000

As React developers, we're always on the lookout for ways to elevate user experience. In this blog post, we'll explore a fascinating challenge: opening a modal when users click the browser's back button in a React application. While React Router v6 offers powerful navigation tools, addressing this specific scenario requires a unique approach.
Let's dive into the details! 🚀✨

const [confirmModal, setConfirmModal] = useState(false)

First of all we are going to take a State for our Modal, so that we can render our modal.

<CustomModal
  open={confirmModal}
  onClose={closeModal}
  containerStyle={{
    width: "90%",
    maxWidth: "380px",
    minWidth: "380px",
    height: "200px",
    backgroundColor: "#F8F8F8",
  }}
>
  <div
    style={{
      display: "flex",
      width: "100%",
      justifyContent: "center",
      alignItems: "center",
      marginBottom: "20px",
    }}
  >
    <p style={{ fontWeight: "600", fontSize: "16px" }}>
      Confirm cancellation of this transaction?
    </p>
  </div>

  <div className="flex justify-between">
    <div>
      <CustomButton title={"Yes"} onClick={leavePage} />
    </div>
    <div>
      <CustomButton title={"No"} onClick={closeModal} />
    </div>
  </div>
</CustomModal>

We've implemented a straightforward modal, and you can also opt for a regular modal. In my case, I'm using a customModal.

Two functions are required initially to manage the modal. The first one is to close the modal:

const closeModal = () => {
    setConfirmModal(false);
};

We will be using react-router-dom V6,
Initialise it with the following lines:

import {  useNavigate } from "react-router-dom";
const navigate = useNavigate();

We are just simply setting the state to false nothing much,
this function will be used when user clicks on the "YES" Button of our Modal.

const leavePage = () => {
    navigate("/");
};

This function will either push the new URL onto the history stack or replace the current URL

    const urlChangeHandler = (url, replace = false) => {
    const historyMethod = replace ? "replaceState" : "pushState";
    window.history[historyMethod];
  };

Here we are using a arrow function where are making use of a window.history.replaceState method.

In simpler terms, the urlChangeHandler function is like a lookout for changes in the web address (URL). Whenever there's a change, it ensures our application stays updated with the new URL by either adding a new entry to the browser's history or replacing the current one. It's like keeping our app in harmony with what's happening in the browser's address bar.

Now's the time we bring in our javascript functions with useEffect Hook, we cannot use javascript native elements directly in react we need to use them inside the useEffect:

  useEffect(() => {
    urlChangeHandler(window.location.href);
    enterValuesInHiddenInput();

    return () => setConfirmModal(true);
  }, []);

In this code snippet, we are using the urlChangeHandler function to handle the initial URL when the component mounts. We pass the current URL (window.location.href) to this function, effectively setting the URL without adding any extra state. This helps initialise the component's state based on the current URL.

We are calling one more function in the above useEffect, that is enterValuesHiddenInput(),

  const hiddenInputRef = useRef(null);

  const enterValuesInHiddenInput = () => {
    hiddenInputRef.current.onclick;
  };

 <button
  ref={hiddenInputRef}
  style={{visibility: "hidden"}}>
          Test
 </button>

In the above code, we are making use of useRef hook, to run a onclick function to forcefully click the screen and then we are hiding this button.

Additionally, at our clean up function we are setting the state of our modal's state true.

The block of code will trigger the above events whenever user clicks on back button of browser

 useEffect(() => {
    enterValuesInHiddenInput();

    const handlePopState = (event) => {
      event.preventDefault();
      setConfirmModal(true);
    };
    window.addEventListener("popstate", handlePopState);
    window.history.pushState({ modalOpened: false }, "");

    return () => {
      window.removeEventListener("popstate", handlePopState);
    };
  }, [hiddenInputRef, urlChangeHandler]);

The popstate event listener in React is triggered when the browser's history navigation changes. It allows developers to respond to back and forward button clicks, enabling smooth navigation within the application.

In this code, we set up an event listener to detect when the user navigates in the browser. When that happens, we activate the handlePopState function, which sets the confirmModal state to true. This ensures the modal opens when the user interacts with the browser's navigation (e.g., clicks the back button). The cleanup function removes the event listener to prevent issues when the component unmounts.

Now, just call your modal inside

{confirmModal && (
        <CustomModal
          open={confirmModal}
          onClose={closeModal}
          containerStyle={{
            width: "90%",
            maxWidth: "380px",
            minWidth: "380px",
            height: "200px",
            backgroundColor: "#F8F8F8",
          }}
        >
          <div
            style={{
              display: "flex",
              width: "100%",
              justifyContent: "center",
              alignItems: "center",
              marginBottom: "20px",
            }}
          >
            <p style={{ fontWeight: "600", fontSize: "16px" }}>
              Are you sure you want to cancel this transaction?
            </p>
          </div>

          <div className="flex justify-between">
            <div>
              <CustomButton title={"Yes"} onClick={leavePage} />
            </div>
            <div>
              <CustomButton title={"No"} onClick={closeModal} />
            </div>
          </div>
        </CustomModal>
      )}

and with that it's done...

Here's the final Output

NOTE: before clicking on the browser back button user must interact with the ui first even as small as a click on a screen is enough...