DEV Community: ZyVOP

AI at Scale: Cost Cuts, New Tooling, and Growing Governance Rules

ZyVOP — Sat, 01 Aug 2026 16:08:14 +0000

Trend: AI is being forced into production at scale

Across the board, today’s headlines reveal a single narrative: AI is no longer a research curiosity but a commodity that must be affordable, operable, and compliant. Companies are slashing model prices, building orchestration tools, and tightening contribution policies—all to make large‑language models (LLMs) viable for everyday business workloads while limiting legal and security exposure.

Price‑performance race – OpenAI’s GPT‑5.6 pricing cuts

OpenAI announced that its new GPT‑5.6 Luna will cost 80 % less and the balanced Terra 20 % less. The move directly addresses the biggest barrier to large‑scale adoption: compute spend. By delivering “more intelligence per dollar,” OpenAI is courting enterprises that need high‑throughput, multi‑step workflows (e.g., automated ticket triage, code generation) without blowing budgets. The benefit is clear – startups and mid‑market firms can now run high‑quality LLM‑driven services that were previously reserved for the cloud‑giants.

Open‑source models prove they can stay uncensored – DeepSeek distillation

CTGT’s experiment distilling DeepSeek V4 Flash into GPT‑OSS‑120B shows that a model trained on outputs of a heavily censored Chinese system can retain financial‑reasoning performance without inheriting the same censorship. This demonstrates that open‑source “teacher‑student” pipelines can decouple capability from geopolitical bias, a crucial insight for companies that need transparent, auditable models while avoiding export‑control entanglements.

Tooling for agents – Agent‑Manager, Grafana AI SDK, Gemini Robotics 2

Operationalizing agents requires orchestration. The Agent‑Manager TUI lets engineers run Claude, Codex, and other agents side‑by‑side in tmux, providing live status, cost gauges, and diff‑based reviews. Similarly, Grafana’s AI SDK for Go offers a wire‑compatible streaming and tool‑calling layer that bridges Go back‑ends with React front‑ends, lowering the engineering effort to embed LLMs in production services. DeepMind’s Gemini Robotics 2 pushes the same orchestration principle into robotics, giving whole‑body intelligence a unified API. Together these projects signal that the ecosystem is maturing from ad‑hoc scripts to disciplined, observable pipelines.

Governance tightening – GCC, OpenJDK, EU platform rules

At the same time, foundational projects are drawing hard lines. The GCC steering committee adopted an AI contributions policy that rejects any “legally significant” LLM‑generated code, effectively banning large‑scale patch contributions. OpenJDK issued an interim policy with identical language, allowing private AI‑assisted debugging but prohibiting AI‑generated commits. In Europe, the EU’s Digital Services Act is being extended to platforms like ChatGPT and Roblox, imposing stricter transparency and safety obligations. The common thread is risk mitigation: intellectual‑property exposure, reviewer fatigue, and geopolitical bias are now seen as operational liabilities that must be codified.

Reality check – Autonomous business experiment fails

Even with these advances, the autonomous‑agent hype remains unproven. Bottleneck Labs gave GPT‑5.6 Sol a real‑world startup challenge and the agent lost $447, generated zero revenue, and only marginally grew its user base. The failure underscores that cost‑effective models and sophisticated tooling are necessary but not sufficient; robust business logic, alignment, and risk‑aware governance are still missing.

What changes next?

We can expect three converging forces:

Price pressure will intensify. Competitors (Anthropic, Google DeepMind) will match or beat OpenAI’s discounting, forcing a race to the bottom that benefits high‑volume SaaS users.
Tooling will become standardized. SDKs like Grafana’s and orchestration UIs will evolve into de‑facto platforms, reducing the friction of deploying multi‑step agents at scale.
Governance will harden. More open‑source foundations will adopt contribution bans, and regulators will demand audit trails for any LLM‑generated code shipped to production.

Companies that align their engineering pipelines with these realities—by adopting cost‑aware models, integrating proven orchestration layers, and embedding compliance checks—will capture the emerging “production AI” market. Those that ignore the policy wave or rely on untested autonomous agents risk both financial loss and legal exposure.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Scrapling: The Python Scraper That Doesn't Break When Sites Change (2026)

ZyVOP — Fri, 31 Jul 2026 10:06:33 +0000

The Problem Every Python Scraper Has

You spend a Tuesday afternoon writing a scraper. CSS selectors, BeautifulSoup, a loop, a CSV export. It works perfectly.

You come back three weeks later. The data is missing. The site redesigned. The class .product-price is now .price-display. Your soup.select_one(".product-price") returns None across all 50,000 URLs you were monitoring.

This is not an edge case. It is the normal lifecycle of every production scraper. D4Vinci/Scrapling — Python web scraper that adapts when sites change, so your data collection keeps working.

Scrapling is the answer to this problem. Instead of finding elements by their exact CSS class — which breaks the moment a developer renames it — Scrapling finds elements by their structural position and visual role in the document. The class name is a hint, not the whole answer. When the hint changes, Scrapling adapts.

By GitHub stars: Firecrawl (131k+), Playwright (90.2k+), Crawl4AI (68.2k+), Scrapy (62.1k+). Scrapling has been growing fast precisely because it solves the problem that every scraper maintainer hates most.

This guide shows you the whole library — the parsing layer (Adaptor), the killer adaptive features (find_similar, auto-generated selectors), and how Scrapling's fetchers handle real websites.

Installation

\n\n

bash\npip install scrapling curl_cffi \n

\n\n

curl_cffi is required by Scrapling's fetchers for TLS fingerprint impersonation. Install both together.

Part 1: The Adaptor — Scrapling's HTML Parser

The Adaptor is Scrapling's HTML parsing layer. Think of it as BeautifulSoup with a smarter element model and additional methods for adaptive scraping.

\n\n

```python\nfrom scrapling.parser import Adaptor

html = """

Senior Python Developer

Infosys
Bangalore
Rs. 18–25 LPA
Python
Django
Apply Now

Data Scientist

TCS
Hyderabad
Rs. 15–22 LPA
Python
ML
Apply Now

"""

Create an Adaptor — pass the HTML and the source URL

The URL is used for resolving relative links and for the auto-trained model

page = Adaptor(html, url="https://techjobs.example.com")
\n```

\n\n

Selecting elements

Scrapling supports CSS selectors and XPath:

\n\n

```python\n# CSS selectors — returns a Selectors list
titles = page.css("h3.job-title")
print([t.text for t in titles])

['Senior Python Developer', 'Data Scientist']

XPath — same return type

titles_xpath = page.xpath("//h3[@class='job-title']")
print([t.text for t in titles_xpath])

['Senior Python Developer', 'Data Scientist']

find_all by tag — like BeautifulSoup's find_all

items = page.find_all("span")
print(f"Found {len(items)} spans")
\n```

\n\n

Getting values out

\n\n

```python\n# .text — the element's visible text content
title_el = page.css("h3.job-title")[0]
print(title_el.text) # Senior Python Developer

.html — the element's outer HTML

print(title_el.html) #

Senior Python Developer

.attrib — dict-like attribute access (never raises KeyError)

card = page.css("div.job-card")[0]
print(card.attrib["data-id"]) # 101
print(card.attrib.get("data-missing")) # None — safe access

Link href

link = page.css("a.apply-btn")[0]
print(link.attrib["href"]) # /jobs/101

All text in a container (concatenates all descendant text nodes)

card_text = card.get_all_text(separator=" ")
print(card_text[:60]) # Senior Python Developer Infosys Bangalore Rs. 18...
\n```

\n\n

extract() and extract_first()

These return the raw HTML string of matched elements — useful when you need to re-parse a sub-section:

\n\n

```python\n# extract_first() → the first match as an HTML string
first_title_html = page.css("h3.job-title").extract_first()
print(first_title_html) #

Senior Python Developer

extract() → list of all matches as HTML strings

all_price_html = page.css("span.salary").extract()
print(all_price_html)

['Rs. 18–25 LPA',

'Rs. 15–22 LPA']

\n```

\n\n

Nested selection

You can run CSS selectors on individual elements, not just the whole page:

\n\n

```python\njobs = []
for card in page.css("div.job-card"):
job = {
"id": card.attrib.get("data-id"),
"title": card.css("h3.job-title")[0].text,
"company": card.css("span.company")[0].text,
"location": card.css("span.location")[0].text,
"salary": card.css("span.salary")[0].text,
"tags": [t.text for t in card.css("span.tag")],
"link": card.css("a.apply-btn")[0].attrib["href"],
}
jobs.append(job)

for j in jobs:
print(f" [{j['id']}] {j['title']} @ {j['company']} — {j['salary']}")
print(f" {', '.join(j['tags'])}")
\n```

\n\n

Traversal

\n\n

```python\n# children — iterate direct child elements
ul_html = "

Item A
Item B
Item C

"
page2 = Adaptor(ul_html, url="https://example.com")
ul = page2.css("ul")[0]

for child in ul.children:
if hasattr(child, "text"):
print(child.text) # Item A, Item B, Item C

find_ancestor — walk up the tree with a test function

html3 = "

Alice

"
page3 = Adaptor(html3, url="https://example.com")
td = page3.css("td.name")[0]

tr = td.find_ancestor(lambda el: el.tag == "tr")
print(tr.tag) # tr
\n```

\n\n

Part 2: The Features That Make Scrapling Different

find_similar() — the adaptive core

This is the feature that makes Scrapling special. find_similar() finds all elements on the page that are structurally similar to a given element — same depth, same relationship to siblings and parent, same content type — even if their CSS class has changed.

\n\n

```python\nhtml = """

Alice	30
Bob	25
Carol	35

"""
page = Adaptor(html, url="https://example.com")

Get only the FIRST name cell by selector

first_name = page.css("td.name")[0]
print("First:", first_name.text) # Alice

find_similar() finds the other name-like cells by structural position

— NOT by class name. If "td.name" becomes "td.member-name" tomorrow,

this still works.

similar = first_name.find_similar()
print("Similar:", [s.text for s in similar]) # ['Bob', 'Carol']
\n```

\n\n

The practical workflow this enables: identify the first example of the element you want, then call find_similar() to get all others. If the site redesigns and changes class names, find_similar() still finds them by structural position.

find_by_text() — locate elements by content

\n\n

```python\nhtml = """

Product Overview

A quality product designed for professionals.

Buy Now

"""
page = Adaptor(html, url="https://example.com")

Find an element by its exact text

btn = page.find_by_text("Buy Now")
print(btn.text) # Buy Now
print(btn.tag) # button

Find by partial text

desc = page.find_by_text("quality product", partial=True)
print(desc.text) # A quality product designed for professionals.
\n```

\n\n

Auto-generated selectors — never write a selector manually again

Scrapling can generate a CSS selector or XPath expression for any element it finds. Use this to discover selectors rather than hunting through DevTools:

\n\n

```python\nhtml = """

"""
page = Adaptor(html, url="https://example.com")

pricing_link = page.find_by_text("Pricing")
print("CSS selector:", pricing_link.generate_css_selector)

body > nav > ul > li:nth-child(2) > a

print("XPath:", pricing_link.generate_xpath_selector)

//body/nav/ul/li[2]/a

print("Full CSS:", pricing_link.generate_full_css_selector)

html > body > nav > ul > li:nth-child(2) > a

\n```

\n\n

Use case: you're scraping a new site and aren't sure what selectors to use. Run find_by_text() on known content, then call generate_css_selector to get the precise selector to use in your production code.

Part 3: Fetching Pages with Scrapling

The Adaptor alone is a parser — you bring the HTML. Scrapling also ships three fetcher classes that handle the HTTP layer and return Adaptor objects directly.

Fetcher — fast HTTP, no browser

Fetcher uses httpx under the hood. Good for simple server-rendered pages with no anti-bot protection.

\n\n

```python\nfrom scrapling.fetchers import Fetcher

fetcher = Fetcher(auto_match=True) # auto_match enables adaptive matching

fetch() returns an Adaptor object — parse it immediately

page = fetcher.get(
"https://books.toscrape.com/",
timeout=20,
stealthy_headers=True, # Add realistic browser headers automatically
)

Use Adaptor methods on the result

titles = page.css("h3 a")
for t in titles[:5]:
print(t.attrib.get("title", t.text))
\n```

\n\n

StealthyFetcher — TLS impersonation

StealthyFetcher uses curl_cffi to impersonate real browser TLS fingerprints. Use this for sites that check JA3/TLS fingerprints (Cloudflare, most major e-commerce sites).

\n\n

```python\nfrom scrapling.fetchers import StealthyFetcher

fetcher = StealthyFetcher(auto_match=True)

page = fetcher.get(
"https://target-site.com/products",
timeout=30,
# Scrapling picks a Chrome impersonation target automatically
)

Result is still an Adaptor — same API

prices = page.css(".price")
print([p.text for p in prices])
\n```

\n\n

PlayWrightFetcher — full browser rendering

For JavaScript-heavy sites:

\n\n

```python\nfrom scrapling.fetchers import PlayWrightFetcher

fetcher = PlayWrightFetcher(auto_match=True)

page = fetcher.get(
"https://spa-site.com/dashboard",
headless=True,
network_idle=True, # Wait for all XHR to complete
timeout=30,
disable_resources=True, # Block images/fonts for speed
)

Same Adaptor API

data = page.css(".data-table tr")
\n```

\n\n

Part 4: The auto_match System

auto_match=True is Scrapling's most powerful feature — and the one that requires a bit of explanation.

When you use auto_match, Scrapling stores a fingerprint of each element you extract — not just its CSS class, but its position in the DOM tree, its relationship to sibling elements, its text content pattern, and other structural signals. On subsequent runs, if the CSS class has changed, Scrapling uses this fingerprint to locate the element anyway.

\n\n

```python\nfrom scrapling.fetchers import Fetcher

fetcher = Fetcher(auto_match=True)

First run: Scrapling fetches the page and learns element fingerprints

page = fetcher.get("https://example.com/products")

You select by CSS class — Scrapling memorises the element

price = page.css_first(".product-price")
print(price.text) # Rs. 1,299

Three weeks later: site redesigns, ".product-price" → ".price-tag"

Scrapling's auto_match finds the element by structural fingerprint

Even though the CSS class changed, it still returns the right element

page2 = fetcher.get("https://example.com/products")
price2 = page2.css_first(".product-price") # Old selector still works!
print(price2.text) # Rs. 1,299 — correct even after the redesign
\n```

\n\n

This is the real promise of Scrapling: selectors that survive redesigns.

Part 5: A Complete Production Scraper with Scrapling

Let's build a complete working scraper that uses Scrapling's full feature set:

\n\n

```python\n# scrapling_full_example.py
import asyncio
import pandas as pd
from scrapling.parser import Adaptor

We use Adaptor with pre-fetched HTML for this example

In production, replace with StealthyFetcher.get() or PlayWrightFetcher.get()

SAMPLE_PAGE_HTML = """<!DOCTYPE html>

Tech Products — Store

Latest Products

<article class="product-card" data-sku="WGT-001">
  <h2 class="product-name">Wireless Keyboard Pro</h2>
  <div class="pricing">
    <span class="current-price">Rs. 3,499</span>
    <span class="original-price">Rs. 4,999</span>
    <span class="discount-badge">30% off</span>
  </div>
  <ul class="feature-list">
    <li>Bluetooth 5.2</li>
    <li>Multi-device pairing</li>
    <li>12-month battery</li>
  </ul>
  <div class="meta">
    <span class="brand">Logitech</span>
    <span class="rating">4.6 ★</span>
    <span class="reviews">2,841 reviews</span>
  </div>
  <a href="/products/WGT-001" class="product-link">View Product</a>
</article>

<article class="product-card" data-sku="MNT-002">
  <h2 class="product-name">4K USB-C Monitor</h2>
  <div class="pricing">
    <span class="current-price">Rs. 28,999</span>
    <span class="original-price">Rs. 34,999</span>
    <span class="discount-badge">17% off</span>
  </div>
  <ul class="feature-list">
    <li>4K 27-inch IPS</li>
    <li>USB-C 90W PD</li>
    <li>HDMI 2.1</li>
  </ul>
  <div class="meta">
    <span class="brand">LG</span>
    <span class="rating">4.8 ★</span>
    <span class="reviews">1,204 reviews</span>
  </div>
  <a href="/products/MNT-002" class="product-link">View Product</a>
</article>

<article class="product-card" data-sku="HDS-003">
  <h2 class="product-name">ANC Headphones Elite</h2>
  <div class="pricing">
    <span class="current-price">Rs. 12,499</span>
    <span class="original-price">Rs. 15,999</span>
    <span class="discount-badge">22% off</span>
  </div>
  <ul class="feature-list">
    <li>Active Noise Cancellation</li>
    <li>40hr battery</li>
    <li>Hi-Res Audio</li>
  </ul>
  <div class="meta">
    <span class="brand">Sony</span>
    <span class="rating">4.7 ★</span>
    <span class="reviews">5,632 reviews</span>
  </div>
  <a href="/products/HDS-003" class="product-link">View Product</a>
</article>

"""

def scrape_products(html: str, source_url: str) -> list[dict]:
"""
Parse product cards from a page using Scrapling.
Demonstrates: nested css(), .text, .attrib, find_similar(), find_by_text()
"""
page = Adaptor(html, url=source_url)
products = []

for card in page.css("article.product-card"):
    sku        = card.attrib.get("data-sku", "")
    name       = card.css("h2.product-name")[0].text

    # Pricing block
    price_curr = card.css("span.current-price")[0].text
    price_orig = card.css("span.original-price")[0].text
    discount   = card.css("span.discount-badge")[0].text

    # Features — list items
    features = [li.text for li in card.css("ul.feature-list li")]

    # Meta
    brand    = card.css("span.brand")[0].text
    rating   = card.css("span.rating")[0].text
    reviews  = card.css("span.reviews")[0].text

    # Link
    link     = card.css("a.product-link")[0].attrib["href"]

    # Clean price to float
    import re
    price_float = float(re.sub(r"[^\d]", "", price_curr))

    products.append({
        "sku":           sku,
        "name":          name,
        "price":         price_float,
        "price_display": price_curr,
        "original":      price_orig,
        "discount":      discount,
        "features":      ", ".join(features),
        "brand":         brand,
        "rating":        rating,
        "reviews":       reviews,
        "url":           link,
    })

return products

def demonstrate_adaptive_features(html: str):
"""
Show Scrapling's key adaptive features:
find_similar(), find_by_text(), auto-generated selectors
"""
page = Adaptor(html, url="https://store.example.com")

print("\n── find_similar() ──")
# Get the first product name
first_name = page.css("h2.product-name")[0]
print(f"First product: {first_name.text}")

# find_similar finds the other product names by structural position
# This works even after CSS class renames
similar = first_name.find_similar()
print(f"Similar elements found: {[s.text for s in similar]}")

print("\n── find_by_text() ──")
# Locate element by its text content — class-independent
sony_brand = page.find_by_text("Sony")
if sony_brand:
    print(f"Found 'Sony' in: <{sony_brand.tag}> element")
    # Walk up to get the parent product card
    parent_card = sony_brand.find_ancestor(lambda el: el.tag == "article")
    if parent_card:
        product_name = parent_card.css("h2")[0].text
        print(f"  → Parent product: {product_name}")

print("\n── find_by_text() partial match ──")
anc_el = page.find_by_text("Noise Cancellation", partial=True)
if anc_el:
    print(f"Partial match: '{anc_el.text}'")

print("\n── Auto-generated selectors ──")
# Let Scrapling generate the selector for any element
first_discount = page.css("span.discount-badge")[0]
print(f"CSS selector: {first_discount.generate_css_selector}")
print(f"XPath:        {first_discount.generate_xpath_selector}")

print("\n── get_all_text() ──")
first_card = page.css("article.product-card")[0]
all_text   = first_card.get_all_text(separator=" | ")
print(f"All text in first card: {all_text[:120]}...")

def main():
print("=" * 55)
print(" SCRAPLING DEMO")
print("=" * 55)

# Scrape products
products = scrape_products(SAMPLE_PAGE_HTML, "https://store.example.com")

print(f"\nScraped {len(products)} products:\n")
df = pd.DataFrame(products)
print(df[["sku", "name", "price", "discount", "brand", "rating"]].to_string(index=False))

# Analysis
print(f"\nAverage price: Rs. {df['price'].mean():,.0f}")
print(f"Most expensive: {df.loc[df['price'].idxmax(), 'name']} "
      f"(Rs. {df['price'].max():,.0f})")
print(f"Cheapest: {df.loc[df['price'].idxmin(), 'name']} "
      f"(Rs. {df['price'].min():,.0f})")

# Demonstrate adaptive features
demonstrate_adaptive_features(SAMPLE_PAGE_HTML)

# Save
df.to_csv("scrapling_products.csv", index=False)
print("\n✓ Data saved to scrapling_products.csv")

if name == "main":
main()
\n```

\n\n

Output:

\n\n

  SCRAPLING DEMO
=======================================================

Scraped 3 products:

     sku                    name     price discount    brand  rating
 WGT-001    Wireless Keyboard Pro    3499.0   30% off Logitech  4.6 ★
 MNT-002          4K USB-C Monitor  28999.0   17% off       LG  4.8 ★
 HDS-003     ANC Headphones Elite   12499.0   22% off     Sony  4.7 ★

Average price: Rs. 14,999
Most expensive: 4K USB-C Monitor (Rs. 28,999)
Cheapest: Wireless Keyboard Pro (Rs. 3,499)

── find_similar() ──
First product: Wireless Keyboard Pro
Similar elements found: ['4K USB-C Monitor', 'ANC Headphones Elite']

── find_by_text() ──
Found 'Sony' in: <span> element
  → Parent product: ANC Headphones Elite

── find_by_text() partial match ──
Partial match: 'Active Noise Cancellation'

── Auto-generated selectors ──
CSS selector: body > div > article > div > span:nth-child(3)
XPath:        //body/div/article/div/span[3]

── get_all_text() ──
All text in first card: Wireless Keyboard Pro | Rs. 3,499 | Rs. 4,999 | 30% off...

✓ Data saved to scrapling_products.csv
\n```

\n\n

* * *

## Part 6: Scrapling vs BeautifulSoup — Honest Comparison

| Feature | BeautifulSoup | Scrapling |
| --- | --- | --- |
| Syntax simplicity | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| CSS selectors | ✅ | ✅ |
| XPath | ❌ (needs lxml directly) | ✅ |
| find_by_text | Basic | Advanced with partial match |
| find_similar | ❌ | ✅ (unique to Scrapling) |
| Auto-selectors | ❌ | ✅ |
| Auto-match on redesign | ❌ | ✅ |
| Built-in fetchers | ❌ | ✅ (3 fetcher types) |
| TLS impersonation | ❌ | ✅ (via StealthyFetcher) |
| Speed | Fast | Slightly slower |
| Documentation | Excellent | Good, growing |
| Community | Enormous | Rapidly growing |

**When to use BeautifulSoup:** You're scraping a stable site, you're a beginner, or you're teaching someone else.

**When to use Scrapling:** You're scraping sites that change frequently, you want built-in fetchers with anti-detection, or you're tired of maintaining selector lists.

* * *

## Part 7: Working with Scrapling's Fetchers in Production

\n\n

```python\n# production_fetcher_example.py
from scrapling.fetchers import StealthyFetcher
import time
import random

def scrape_with_retry(url: str, max_retries: int = 3) -> object | None:
    """
    Production-safe wrapper around StealthyFetcher.
    Handles retries and errors gracefully.
    """
    fetcher = StealthyFetcher(auto_match=True)

    for attempt in range(max_retries):
        try:
            page = fetcher.get(
                url,
                timeout        = 25,
                stealthy_headers=True,    # Scrapling adds realistic headers
                hide_browser   = True,    # Additional stealth for PlayWright
            )
            return page

        except Exception as e:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"  Attempt {attempt+1} failed: {e}. Retrying in {wait:.1f}s")
            time.sleep(wait)

    print(f"  Permanently failed: {url}")
    return None

# Example multi-page scrape
def scrape_catalog(base_url: str, pages: int = 5) -> list[dict]:
    all_products = []

    for page_num in range(1, pages + 1):
        url  = f"{base_url}?page={page_num}"
        page = scrape_with_retry(url)

        if not page:
            continue

        # Adaptive scraping — if class names change between pages,
        # find_similar() on first result finds all counterparts
        first_price = page.css(".price, .product-price, .current-price")
        if not first_price:
            print(f"  No prices found on page {page_num}")
            continue

        # find_similar adapts to whatever the price class is on this page
        all_prices = first_price[0].find_similar()
        all_names  = page.css("h2, h3").extract()

        print(f"  Page {page_num}: {len(all_prices)} prices, {len(all_names)} names")
        time.sleep(random.uniform(1.5, 3.0))

    return all_products
\n```

\n\n

* * *

## Common Pitfalls and Fixes

`css_first()` **doesn't exist — use** `css()[0]`

\n\n

```python\n# Wrong — AttributeError
item = page.css_first(".price")

# Right
item = page.css(".price")[0] if page.css(".price") else None
\n```

\n\n

`extract()` **returns HTML strings, not text**

\n\n

```python\n# extract() returns '<span class="price">Rs. 999</span>'
raw = page.css(".price").extract()

# For clean text, iterate and use .text
texts = [el.text for el in page.css(".price")]
\n```

\n\n

`find_ancestor()` **takes a callable, not a string**

\n\n

```python\n# Wrong
parent = el.find_ancestor("div")

# Right
parent = el.find_ancestor(lambda e: e.tag == "div")
parent = el.find_ancestor(lambda e: "card" in (e.attrib.get("class") or ""))
\n```

\n\n

* * *

## Summary

| Concept | Method | Example |
| --- | --- | --- |
| Parse HTML | Adaptor(html, url=url) | Build the page object |
| CSS select all | page.css("selector") | Returns Selectors list |
| CSS select one | page.css("sel")[0] | First match |
| Element text | el.text | Clean inner text |
| Attribute | el.attrib["href"] | Safe dict access |
| All text | el.get_all_text(sep) | Concatenated descendants |
| Tag name | el.tag | "div", "span", etc. |
| Outer HTML | el.html | &lt;div&gt;...&lt;/div&gt; |
| XPath | page.xpath("//h2") | Same return type as css() |
| Find by text | page.find_by_text("text") | Class-independent search |
| Partial match | find_by_text("x", partial=True) | Substring search |
| Find similar | el.find_similar() | Adaptive structural match |
| Walk up tree | el.find_ancestor(lambda e: ...) | Callable test |
| Auto CSS | el.generate_css_selector | Discover selectors |
| Auto XPath | el.generate_xpath_selector | Discover selectors |
| Children | el.children | Direct child iterator |
| Raw HTML list | page.css("x").extract() | HTML strings |
| First raw HTML | page.css("x").extract_first() | First HTML string |

* * *

---

*Originally published on [ZyVOP](https://zyvop.com/scrapling-the-python-scraper-that-doesn-t-break-when-sites-change-2026-igvn0)*

💡 For more articles like this, [subscribe to the ZyVOP newsletter](https://zyvop.com/newsletter)!

We Added Redis to Make Things Faster. It Made Things Worse.

ZyVOP — Thu, 30 Jul 2026 10:02:44 +0000

A product endpoint is taking around 800ms under load.

The request isn't doing anything unusual: fetch a product from PostgreSQL, serialize it, return JSON. The data changes infrequently, but the same query runs thousands of times.

Redis looks like an obvious fix.

\n\n

``ts\nasync function getProduct(id: string) { const key =product:${id}`;

const cached = await redis.get(key);

if (cached) {
return JSON.parse(cached);
}

const product = await productRepository.findOneBy({ id });

if (product) {
await redis.set(key, JSON.stringify(product), {
EX: 3600,
});
}

return product;
}
\n```

\n\n

On a cache hit, an expensive database read disappears from the request path.

For example:

\n\n

text\nDatabase read: ~800ms Redis hit: ~30ms \n

\n\n

Those numbers are illustrative, but the improvement can be substantial when the original operation is expensive.

So far, Redis is doing exactly what we wanted.

The interesting problems start when the underlying data changes.

The Database Is Correct. The Response Isn't.

Consider product 123.

PostgreSQL contains:

\n\n

  "id": "123",
  "name": "Mechanical Keyboard",
  "price": 129
}
\n```

\n\n

Redis still contains an older copy:

\n\n

```json\n{
  "id": "123",
  "name": "Mechanical Keyboard",
  "price": 99
}
\n```

\n\n

The application checks Redis first:

\n\n

```ts\nconst cached = await redis.get("product:123");

if (cached) {
  return JSON.parse(cached);
}
\n```

\n\n

Nothing fails.

The database is healthy. Redis is healthy. The request returns `200 OK`.

It also returns the wrong price.

The obvious fix is invalidation.

\n\n

```ts\nawait productRepository.save(product);
await redis.del(`product:${product.id}`);
\n```

\n\n

For a single cache key, that's straightforward.

Most applications don't have a single cache key.

## One Row Can Exist in Several Cached Responses

Product `123` might appear in:

\n\n

```text\nproduct:123
products:list
products:category:keyboards
products:featured
search:mechanical-keyboard
\n```

\n\n

Updating one database row can make every one of those entries stale.

The write path now needs to know about the read paths:

\n\n

```ts\nawait productRepository.save(product);

await Promise.all([
  redis.del(`product:${product.id}`),
  redis.del("products:list"),
  redis.del(`products:category:${product.categoryId}`),
  redis.del("products:featured"),
]);
\n```

\n\n

And that still doesn't necessarily handle search results, filtered lists, pagination, or other cached representations containing the product.

This is where caching starts affecting application design.

Without caching:

\n\n

```text\nupdate product -> database
\n```

\n\n

With several cached representations:

\n\n

```text\nupdate product
      |
      +--> database
      +--> product cache
      +--> category cache
      +--> list cache
      +--> featured cache
      +--> search cache
\n```

\n\n

Reads became cheaper, but correctness now depends on invalidating every relevant representation.

The more places the same data is cached, the harder that becomes.

## Redis Failure Is Not the Same as a Cache Miss

A common cache-aside implementation assumes this:

\n\n

```ts\nconst cached = await redis.get(key);

if (cached) {
  return JSON.parse(cached);
}

return database.findProduct(id);
\n```

\n\n

But if Redis is unavailable, `redis.get()` doesn't necessarily return `null`.

It may throw.

It may also spend time reconnecting or waiting for a network timeout before failing.

So a real fallback needs to treat Redis as a dependency that can fail.

\n\n

```ts\nasync function getProduct(id: string) {
  const key = `product:${id}`;

  try {
    const cached = await redis.get(key);

    if (cached) {
      return JSON.parse(cached);
    }
  } catch (error) {
    logger.warn({ error, key }, "Redis read failed");
  }

  const product = await productRepository.findOneBy({ id });

  if (!product) {
    return null;
  }

  try {
    await redis.set(key, JSON.stringify(product), {
      EX: 3600,
    });
  } catch (error) {
    logger.warn({ error, key }, "Redis write failed");
  }

  return product;
}
\n```

\n\n

This allows the database path to continue when Redis fails.

It doesn't mean the system is safe.

Suppose Redis normally absorbs 90% of product reads.

\n\n

```text\n10,000 requests/minute

Redis hits:        9,000
PostgreSQL reads:  1,000
\n```

\n\n

Now Redis becomes unavailable.

\n\n

```text\n10,000 requests/minute

Redis hits:            0
PostgreSQL reads: 10,000
\n```

\n\n

The database didn't fail.

Its workload changed by an order of magnitude.

\n\n

```mermaid\nflowchart TD
    A[Redis unavailable] --> B[Requests fall back to PostgreSQL]
    B --> C[Database traffic increases]
    C --> D[Connection pool fills]
    D --> E[Queries wait longer]
    E --> F[Requests remain open longer]
    F --> G[Application concurrency increases]
\n```

\n\n

This is why testing a Redis outage with one request proves very little.

The interesting test is Redis unavailable **under normal production traffic**.

There is another detail worth handling: the Redis client itself needs sensible connection and command timeouts. A fallback doesn't help much if every request waits several seconds for Redis before reaching PostgreSQL.

## TTLs Can Move Load Instead of Removing It

Suppose thousands of cache entries are populated during a batch import or deployment.

They all receive the same TTL:

\n\n

```ts\nconst CACHE_TTL = 3600;

await redis.set(key, value, {
  EX: CACHE_TTL,
});
\n```

\n\n

If many entries are created around 10:00, many become eligible for expiration around 11:00.

Traffic that was previously hitting Redis starts rebuilding those entries from the database.

Instead of database work being spread across the hour, some of it becomes concentrated around expiration.

A small amount of TTL jitter helps avoid unnecessary synchronization:

\n\n

```ts\nfunction ttlWithJitter(baseSeconds: number) {
  const jitter = Math.floor(Math.random() * 300);

  return baseSeconds + jitter;
}

await redis.set(key, value, {
  EX: ttlWithJitter(3600),
});
\n```

\n\n

Entries created together no longer have identical expiration times.

Jitter doesn't solve cache stampedes, though.

## One Expired Key Can Trigger Many Queries

Consider a popular cache entry:

\n\n

```text\nhomepage:products
\n```

\n\n

It expires.

Before any request has rebuilt it, 100 requests arrive.

Each one executes:

\n\n

```ts\nconst cached = await redis.get(key);

if (!cached) {
  return loadFromDatabase();
}
\n```

\n\n

All 100 requests observe the same miss.

\n\n

```mermaid\nflowchart TD
    A[homepage:products expires] --> B[Request 1]
    A --> C[Request 2]
    A --> D[Request 3]
    A --> E[Request 4...]

    B --> F[Expensive database query]
    C --> F
    D --> F
    E --> F
\n```

\n\n

The cache normally prevents repeated work, but during that window it provides no coordination between callers.

This is a cache stampede.

For a single Node.js process, duplicate work can be coalesced with an in-flight promise:

\n\n

```ts\nconst pending = new Map<string, Promise<unknown>>();

async function loadOnce<T>(
  key: string,
  loader: () => Promise<T>,
): Promise<T> {
  const existing = pending.get(key);

  if (existing) {
    return existing as Promise<T>;
  }

  const request = loader().finally(() => {
    pending.delete(key);
  });

  pending.set(key, request);

  return request;
}
\n```

\n\n

Now concurrent requests inside that process can share one rebuild.

But process-local coordination has an important limitation.

Suppose the application runs four instances:

\n\n

```text\n             Load Balancer
          /       |       \
       API-1    API-2    API-3    API-4
         |        |        |        |
         +--------+--------+--------+
                          |
                     PostgreSQL
\n```

\n\n

Each instance has its own `pending` map.

The same expired key can therefore trigger one rebuild per instance.

Four instances may produce four expensive queries instead of 100. That's much better, but it isn't globally coordinated.

For expensive or high-traffic keys, options include distributed locking, stale-while-revalidate, background refresh, or other forms of cross-instance request coordination.

Which one makes sense depends on how expensive stale data and duplicate work are for that particular cache.

## Sometimes Redis Is Hiding the Real Problem

Suppose a database query takes 400ms.

Caching it might reduce most requests to a few milliseconds.

Before adding Redis, run:

\n\n

```sql\nEXPLAIN ANALYZE
SELECT ...
\n```

\n\n

Maybe the query is scanning 800,000 rows because an index is missing.

After fixing the query:

\n\n

```text\nBefore index: ~400ms
After index:   ~30ms
\n```

\n\n

Those numbers are an example, but the point matters: the cache would have hidden a query that should have been fixed.

The same applies to N+1 queries.

Or fetching entire relations when the response needs three fields.

Or repeatedly calling an external service when the data could have been included in the original request.

Caching a slow operation and optimizing a slow operation are different things.

Before deciding to cache an endpoint, it helps to know why the endpoint is expensive.

## Not Everything Expensive Should Be Cached

A useful cache candidate usually has some combination of:

-   expensive computation or I/O;

-   frequent reads;

-   relatively infrequent changes;

-   tolerance for some amount of staleness;

-   predictable invalidation.


A public category list might fit well.

\n\n

```text\ncategories:all
\n```

\n\n

It is requested frequently and probably changes infrequently.

A user dashboard is different.

\n\n

```text\ndashboard:user:4821
\n```

\n\n

It might contain:

\n\n

```text\npermissions
subscription state
usage limits
billing information
recent activity
account settings
\n```

\n\n

Some of those values can change independently, and some may need stronger freshness guarantees than others.

Caching the entire dashboard as one object may save database work, but it also turns several independent data sources into one invalidation problem.

The question isn't only whether caching makes the endpoint faster.

The question is whether the performance gain is worth the new correctness problem.

## Cache Keys Are Part of the Architecture

Cache keys often start simple:

\n\n

```text\nproduct:123
\n```

\n\n

Then requirements arrive.

Different currencies:

\n\n

```text\nproduct:123:USD
product:123:EUR
\n```

\n\n

Different locales:

\n\n

```text\nproduct:123:en
product:123:fr
\n```

\n\n

Different tenants:

\n\n

```text\ntenant:42:product:123
tenant:91:product:123
\n```

\n\n

Different permissions or response variants can add more dimensions.

At that point, cache-key design isn't an implementation detail.

A missing dimension can return stale data.

A missing tenant identifier can return someone else's data.

A key format that is difficult to invalidate can turn a simple update into a broad cache purge.

The key needs to represent every input that can materially change the cached response.

That is easy to say and surprisingly easy to get wrong.

## What a Safer Cache-Aside Path Looks Like

A cache-aside read path is still simple conceptually:

\n\n

```mermaid\nflowchart LR
    A[Request] --> B{Cache appropriate?}

    B -->|No| C[Database]
    B -->|Yes| D[Redis]

    D -->|Hit| E[Return cached value]
    D -->|Miss or failure| C

    C --> F[Build response]
    F --> G[Populate cache]
    F --> H[Return response]
\n```

\n\n

The difficult parts are outside that diagram:

-   deciding what deserves to be cached;

-   choosing keys that represent the actual response;

-   invalidating every affected representation;

-   preventing synchronized expiration;

-   controlling expensive rebuilds;

-   handling Redis latency and failure;

-   ensuring the database can survive reduced cache effectiveness.


Redis solves none of those automatically.

## Redis Wasn't the Problem

Redis can remove an enormous amount of repeated work from a system.

It can also make a poorly understood performance problem harder to see.

If an endpoint is slow, start with the endpoint.

Measure the query.

Look at the query plan.

Check how much data is being fetched.

Check downstream calls.

Look for repeated computation.

Then decide whether the remaining work is something worth caching.

The better question isn't:

> Should this use Redis?

It's:

> What work are we avoiding, how long can we reuse its result, and what happens when that result is stale or unavailable?

If those answers are clear, Redis can be remarkably effective.

If they aren't, a 30ms cache hit may simply be hiding the next production problem.

---

*Originally published on [ZyVOP](https://zyvop.com/we-added-redis-to-make-things-faster-it-made-things-worse-4ganu)*

💡 For more articles like this, [subscribe to the ZyVOP newsletter](https://zyvop.com/newsletter)!

Scraping Social Media Data with Python: X (Twitter), Reddit & Instagram (2026)

ZyVOP — Fri, 24 Jul 2026 11:17:10 +0000

Over 500 million posts are published on X (formerly Twitter) every single day. Reddit hosts more than 100,000 active communities discussing everything from machine learning to local weather. Instagram's 2 billion users generate a continuous stream of product opinions, lifestyle signals, and trend indicators.

Social media data is the closest thing the internet has to real-time public opinion. It powers:

Sentiment analysis — What do people think about your product, brand, or competitor right now?
Trend detection — What topics are accelerating before they go mainstream?
Academic research — Studying misinformation, political discourse, crisis communication
Market intelligence — Tracking competitor mentions, industry conversations
AI training datasets — Social text is gold for fine-tuning conversational models

The challenge in 2026: platforms have made scraping harder than ever. The official X API now starts at $100/month and the free tier is almost useless for research. Instagram aggressively blocks headless browsers. Reddit has rate-limited its API severely after the 2023 controversy.

This guide gives you working approaches for all three platforms — from DIY Python scrapers to managed tools — with honest assessments of what each method can and cannot do.

The 2026 Reality Check: What's Still Scrapable

Before writing a single line of code, here's the honest state of social media scraping in 2026:

Platform	Public Posts	Profile Data	Followers	DMs	Difficulty
X / Twitter	✅ (with effort)	✅	✅	❌	Hard
Reddit	✅ (via API/PRAW)	✅	❌	❌	Easy–Medium
Instagram	⚠️ Public only	⚠️ Public only	❌	❌	Very Hard
TikTok	✅ (via mobile API)	✅	❌	❌	Hard
LinkedIn	⚠️ (see Blog 02)	⚠️ (see Blog 02)	❌	❌	Hard

Key principles:

Never scrape private data, DMs, or anything behind a login you don't own
Always respect robots.txt and Terms of Service for your use case
For academic or commercial research, official APIs or licensed providers are the safer path

Part 1: Scraping X (Twitter) in 2026

The Landscape

X's public guest API was effectively removed in 2023. Almost every endpoint now requires a logged-in session and a CSRF token. The internal GraphQL API that underpins X's own web interface is the main DIY scraping target — but its endpoint identifiers change regularly.

Three working approaches exist in 2026:

Method	Cost	Volume	Reliability	Best for
X API v2 (official)	$100+/month	Limited	High	Authorised developers
Playwright + session	Free	Low-medium	Medium	Research, personal use
ScrapeGraphAI	Paid per req	High	High	Production

Method 1: X API v2 (Official)

For authorised projects, the official API is always the right choice. The free tier gives 500k tweets/month read access:

pip install tweepy

import tweepy
import pandas as pd
from datetime import datetime, timezone, timedelta

# Get credentials at developer.twitter.com
client = tweepy.Client(bearer_token="YOUR_BEARER_TOKEN")

def search_recent_tweets(
    query: str,
    max_results: int = 100,
    days_back: int = 7
) -> pd.DataFrame:
    """
    Search tweets from the last 7 days using X API v2.

    Args:
        query: Search query. Supports operators like:
               - AND/OR: "python scraping OR web crawling"
               - Exact: '"machine learning" tutorial'
               - Exclude: "python -is:retweet"
               - Language: "python lang:en"
               - Has media: "python has:images"
    """
    start_time = datetime.now(timezone.utc) - timedelta(days=days_back)

    # Append filters to reduce noise
    full_query = f"{query} -is:retweet lang:en"

    tweets_data = []
    paginator = tweepy.Paginator(
        client.search_recent_tweets,
        query=full_query,
        tweet_fields=[
            "created_at", "public_metrics", "author_id",
            "lang", "context_annotations", "entities"
        ],
        user_fields=["name", "username", "public_metrics", "verified"],
        expansions=["author_id"],
        start_time=start_time,
        max_results=min(100, max_results),
    )

    users_map = {}

    for response in paginator:
        if not response.data:
            break

        # Build user lookup from includes
        if response.includes and "users" in response.includes:
            for user in response.includes["users"]:
                users_map[user.id] = user

        for tweet in response.data:
            user = users_map.get(tweet.author_id)
            metrics = tweet.public_metrics or {}

            tweets_data.append({
                "tweet_id":       str(tweet.id),
                "text":           tweet.text,
                "created_at":     tweet.created_at,
                "author_id":      str(tweet.author_id),
                "username":       user.username if user else None,
                "name":           user.name if user else None,
                "followers":      user.public_metrics.get("followers_count") if user else None,
                "retweets":       metrics.get("retweet_count", 0),
                "likes":          metrics.get("like_count", 0),
                "replies":        metrics.get("reply_count", 0),
                "quotes":         metrics.get("quote_count", 0),
                "impressions":    metrics.get("impression_count", 0),
                "url":            f"https://x.com/{user.username if user else 'i'}/status/{tweet.id}",
            })

        if len(tweets_data) >= max_results:
            break

    df = pd.DataFrame(tweets_data)
    print(f"Collected {len(df)} tweets for query: '{query}'")
    return df

# Example: track brand mentions
df = search_recent_tweets(
    query="python web scraping 2026",
    max_results=200,
    days_back=7
)
df.to_csv("tweets.csv", index=False)
print(df[["username", "text", "likes", "retweets"]].head(10))

Method 2: Playwright-Based X Scraper (No API Key)

For personal research without API access, Playwright can scrape public search results and profiles using a saved session — exactly as described for LinkedIn in Blog 02:

import asyncio
import random
import json
from playwright.async_api import async_playwright
from playwright_stealth import stealth_async

async def save_x_session():
    """Log in to X manually once and save cookies. Run this first."""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120"
        )
        page = await context.new_page()
        await page.goto("https://x.com/login")

        print("Log in manually, then press Enter...")
        input()

        cookies = await context.cookies()
        with open("x_cookies.json", "w") as f:
            json.dump(cookies, f)
        print(f"Saved {len(cookies)} cookies.")
        await browser.close()

async def scrape_x_search(query: str, scroll_times: int = 10) -> list[dict]:
    """
    Scrape X search results using a saved session.
    Returns a list of tweet dicts.
    """
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=["--disable-blink-features=AutomationControlled"]
        )
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120",
            viewport={"width": 1280, "height": 900}
        )

        # Load saved session
        with open("x_cookies.json") as f:
            await context.add_cookies(json.load(f))

        page = await context.new_page()
        await stealth_async(page)

        # Navigate to search
        encoded = query.replace(" ", "%20")
        await page.goto(
            f"https://x.com/search?q={encoded}&src=typed_query&f=top",
            wait_until="domcontentloaded"
        )
        await asyncio.sleep(random.uniform(2, 4))

        tweets = []
        seen_ids = set()

        for scroll in range(scroll_times):
            # Extract tweet data from the current viewport
            tweet_articles = await page.query_selector_all("article[data-testid='tweet']")

            for article in tweet_articles:
                try:
                    # Get tweet text
                    text_el = await article.query_selector("[data-testid='tweetText']")
                    text = await text_el.inner_text() if text_el else None

                    # Username
                    user_el = await article.query_selector("[data-testid='User-Name'] a")
                    href = await user_el.get_attribute("href") if user_el else ""
                    username = href.strip("/").split("/")[-1] if href else None

                    # Display name
                    name_els = await article.query_selector_all(
                        "[data-testid='User-Name'] span"
                    )
                    display_name = None
                    for el in name_els:
                        txt = await el.inner_text()
                        if txt and not txt.startswith("@"):
                            display_name = txt
                            break

                    # Engagement stats
                    async def get_stat(testid):
                        el = await article.query_selector(f"[data-testid='{testid}']")
                        if el:
                            txt = await el.inner_text()
                            return txt.strip() or "0"
                        return "0"

                    likes    = await get_stat("like")
                    replies  = await get_stat("reply")
                    retweets = await get_stat("retweet")

                    # Time
                    time_el = await article.query_selector("time")
                    posted_at = await time_el.get_attribute("datetime") if time_el else None

                    # Unique ID to deduplicate
                    tweet_id = f"{username}_{posted_at}"
                    if tweet_id in seen_ids or not text:
                        continue
                    seen_ids.add(tweet_id)

                    tweets.append({
                        "username":     username,
                        "display_name": display_name,
                        "text":         text,
                        "likes":        likes,
                        "replies":      replies,
                        "retweets":     retweets,
                        "posted_at":    posted_at,
                    })
                except Exception:
                    continue

            # Scroll down for more tweets
            await page.evaluate("window.scrollBy(0, window.innerHeight * 1.5)")
            await asyncio.sleep(random.uniform(1.5, 3.0))
            print(f"  Scroll {scroll+1}/{scroll_times} — {len(tweets)} tweets collected")

        await browser.close()

    print(f"\nTotal unique tweets: {len(tweets)}")
    return tweets

# Run
tweets = asyncio.run(scrape_x_search("python scraping 2026", scroll_times=8))
df = pd.DataFrame(tweets)
df.to_csv("x_search_results.csv", index=False)

Part 2: Reddit Scraping with PRAW

Reddit is the most developer-friendly major social platform for data collection. Its official Python wrapper PRAW (Python Reddit API Wrapper) is free, well-documented, and gives access to posts, comments, user profiles, and subreddit data at no cost.

pip install praw pandas

Scraping subreddit posts

import praw
import pandas as pd
from datetime import datetime, timezone

reddit = praw.Reddit(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET",
    user_agent="PythonResearchBot/1.0 by u/your_username",
)

def scrape_subreddit(
    subreddit_name: str,
    sort: str = "hot",         # "hot", "new", "top", "rising"
    time_filter: str = "week", # "hour", "day", "week", "month", "year", "all"
    limit: int = 500
) -> pd.DataFrame:
    """
    Scrape posts from a subreddit.

    Best for: topic research, trend monitoring, content analysis.
    """
    sub = reddit.subreddit(subreddit_name)

    # Choose sort method
    if sort == "hot":
        posts_gen = sub.hot(limit=limit)
    elif sort == "new":
        posts_gen = sub.new(limit=limit)
    elif sort == "top":
        posts_gen = sub.top(time_filter=time_filter, limit=limit)
    elif sort == "rising":
        posts_gen = sub.rising(limit=limit)
    else:
        posts_gen = sub.hot(limit=limit)

    records = []
    for post in posts_gen:
        records.append({
            "post_id":       post.id,
            "title":         post.title,
            "text":          post.selftext[:500] if post.selftext else None,
            "author":        str(post.author) if post.author else "[deleted]",
            "score":         post.score,
            "upvote_ratio":  post.upvote_ratio,
            "num_comments":  post.num_comments,
            "url":           post.url,
            "permalink":     f"https://reddit.com{post.permalink}",
            "flair":         post.link_flair_text,
            "is_self":       post.is_self,
            "created_utc":   datetime.fromtimestamp(post.created_utc, tz=timezone.utc).isoformat(),
            "subreddit":     subreddit_name,
        })

    df = pd.DataFrame(records)
    print(f"Scraped {len(df)} posts from r/{subreddit_name}")
    return df

# Example: research Python discussions
df = scrape_subreddit("learnpython", sort="top", time_filter="month", limit=500)
df.to_csv("reddit_learnpython.csv", index=False)
print(df[["title", "score", "num_comments"]].head(10))

Scraping post comments (deep dive)

def scrape_post_comments(
    post_url: str,
    max_comments: int = 200,
    include_replies: bool = False
) -> pd.DataFrame:
    """
    Scrape all comments from a Reddit post.
    Useful for sentiment analysis, topic deep-dives, building datasets.
    """
    submission = reddit.submission(url=post_url)

    # Replace "MoreComments" objects to get all comments
    submission.comments.replace_more(limit=0)

    all_comments = submission.comments.list() if include_replies else submission.comments

    records = []
    for comment in list(all_comments)[:max_comments]:
        if not hasattr(comment, "body"):
            continue
        records.append({
            "comment_id":    comment.id,
            "author":        str(comment.author) if comment.author else "[deleted]",
            "body":          comment.body,
            "score":         comment.score,
            "depth":         comment.depth,
            "created_utc":   datetime.fromtimestamp(
                                 comment.created_utc, tz=timezone.utc
                             ).isoformat(),
            "is_op":         comment.is_submitter,
            "awards":        comment.total_awards_received,
        })

    return pd.DataFrame(records)

# Example: deep dive on a specific post
comments_df = scrape_post_comments(
    "https://www.reddit.com/r/MachineLearning/comments/example/",
    max_comments=300
)
print(f"Top comment: {comments_df.sort_values('score', ascending=False).iloc[0]['body'][:200]}")

Multi-subreddit keyword monitoring

def monitor_keyword_across_subreddits(
    keyword: str,
    subreddits: list[str],
    limit_per_sub: int = 100
) -> pd.DataFrame:
    """
    Search for a keyword across multiple subreddits.
    Great for competitive intelligence and brand monitoring.
    """
    all_posts = []

    for sub_name in subreddits:
        print(f"Searching r/{sub_name} for '{keyword}'...")
        try:
            sub = reddit.subreddit(sub_name)
            for post in sub.search(keyword, sort="new", limit=limit_per_sub):
                all_posts.append({
                    "subreddit":    sub_name,
                    "title":        post.title,
                    "text":         post.selftext[:300] if post.selftext else None,
                    "author":       str(post.author) if post.author else "[deleted]",
                    "score":        post.score,
                    "comments":     post.num_comments,
                    "permalink":    f"https://reddit.com{post.permalink}",
                    "created_utc":  datetime.fromtimestamp(
                                        post.created_utc, tz=timezone.utc
                                    ).isoformat(),
                })
        except Exception as e:
            print(f"  Error on r/{sub_name}: {e}")

    df = pd.DataFrame(all_posts)

    # Summary by subreddit
    summary = df.groupby("subreddit").agg(
        posts=("title", "count"),
        avg_score=("score", "mean"),
        total_comments=("comments", "sum")
    ).sort_values("posts", ascending=False)

    print(f"\n── Results for '{keyword}' ──")
    print(summary.to_string())

    return df

# Track "Python scraping" across tech subreddits
df = monitor_keyword_across_subreddits(
    keyword="python scraping",
    subreddits=["Python", "learnpython", "webdev", "datascience", "MachineLearning"],
    limit_per_sub=50
)
df.to_csv("reddit_keyword_monitor.csv", index=False)

Part 3: Instagram Scraping (Public Profiles Only)

Instagram is the most aggressively protected major platform in 2026. The public guest API was removed years ago. Almost all data requires a logged-in session. The GraphQL API underlying Instagram's web interface changes identifiers constantly.

This section covers only public profile and hashtag data — the bare minimum available without login — and the Playwright approach for logged-in access to your own account's data.

Scraping public profile metadata

import httpx
import json
import asyncio

async def get_instagram_profile(username: str) -> dict | None:
    """
    Fetch public profile data for an Instagram username.
    Uses the semi-public shared data endpoint.
    Only works for public accounts.
    """
    url = f"https://www.instagram.com/{username}/?__a=1&__d=dis"

    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/120 Safari/537.36"
        ),
        "Accept": "*/*",
        "Referer": "https://www.instagram.com/",
        "X-IG-App-ID": "936619743392459",  # Instagram Web app ID
    }

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
        try:
            r = await client.get(url, timeout=15)
            if r.status_code == 200:
                data = r.json()
                user = data.get("graphql", {}).get("user", {})
                return {
                    "username":       user.get("username"),
                    "full_name":      user.get("full_name"),
                    "biography":      user.get("biography"),
                    "followers":      user.get("edge_followed_by", {}).get("count"),
                    "following":      user.get("edge_follow", {}).get("count"),
                    "posts":          user.get("edge_owner_to_timeline_media", {}).get("count"),
                    "is_verified":    user.get("is_verified"),
                    "is_business":    user.get("is_business_account"),
                    "profile_url":    f"https://www.instagram.com/{username}/",
                }
        except Exception as e:
            print(f"Error fetching @{username}: {e}")
    return None

# Batch scrape public profiles
async def batch_profile_scrape(usernames: list[str]) -> pd.DataFrame:
    results = []
    for username in usernames:
        print(f"Fetching @{username}...")
        profile = await get_instagram_profile(username)
        if profile:
            results.append(profile)
        await asyncio.sleep(random.uniform(2, 5))
    return pd.DataFrame(results)

profiles_df = asyncio.run(batch_profile_scrape([
    "natgeo", "nasa", "python.learning"
]))
print(profiles_df[["username", "followers", "posts", "is_verified"]])

Playwright-based Instagram scraper (logged-in session)

For your own account's data or influencer research on public accounts:

import asyncio, json, random
from playwright.async_api import async_playwright
from playwright_stealth import stealth_async

async def save_instagram_session():
    """Log in manually and save session cookies. Run once."""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) "
                       "AppleWebKit/605.1.15 Mobile/15E148 Safari/604.1",
            viewport={"width": 390, "height": 844},
            is_mobile=True,
        )
        page = await context.new_page()
        await page.goto("https://www.instagram.com/accounts/login/")
        print("Log in manually, then press Enter...")
        input()
        cookies = await context.cookies()
        with open("ig_cookies.json", "w") as f:
            json.dump(cookies, f)
        print("Session saved.")
        await browser.close()

async def scrape_instagram_posts(username: str, max_posts: int = 30) -> list[dict]:
    """
    Scrape recent posts from a public Instagram profile.
    Requires a saved session from save_instagram_session().
    """
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=["--no-sandbox", "--disable-blink-features=AutomationControlled"]
        )
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120",
            viewport={"width": 1280, "height": 900}
        )
        with open("ig_cookies.json") as f:
            await context.add_cookies(json.load(f))

        page = await context.new_page()
        await stealth_async(page)

        await page.goto(
            f"https://www.instagram.com/{username}/",
            wait_until="domcontentloaded"
        )
        await asyncio.sleep(random.uniform(2, 4))

        posts = []
        # Scroll to load posts
        for _ in range(max_posts // 12 + 1):
            post_links = await page.query_selector_all("article a[href*='/p/']")
            for link in post_links:
                href = await link.get_attribute("href")
                if href and href not in [p.get("href") for p in posts]:
                    posts.append({"href": href, "username": username})
            await page.evaluate("window.scrollBy(0, window.innerHeight)")
            await asyncio.sleep(random.uniform(1, 2.5))

        # Deduplicate and limit
        seen = set()
        unique_posts = []
        for p in posts:
            if p["href"] not in seen:
                seen.add(p["href"])
                unique_posts.append(p)
        posts = unique_posts[:max_posts]

        await browser.close()

    print(f"Found {len(posts)} posts for @{username}")
    return posts

Part 4: Sentiment Analysis on Collected Data

Once you've collected social media data, analysing sentiment turns raw text into actionable signals. The transformers library from HuggingFace provides excellent pre-trained sentiment models:

pip install transformers torch

from transformers import pipeline
import pandas as pd

# Load a pre-trained sentiment model (downloads ~67MB on first run)
# "cardiffnlp/twitter-roberta-base-sentiment-latest" is trained on tweets
sentiment_pipeline = pipeline(
    "sentiment-analysis",
    model="cardiffnlp/twitter-roberta-base-sentiment-latest",
    tokenizer="cardiffnlp/twitter-roberta-base-sentiment-latest",
    max_length=512,
    truncation=True,
)

LABEL_MAP = {
    "LABEL_0": "negative",
    "LABEL_1": "neutral",
    "LABEL_2": "positive",
}

def analyse_sentiment_batch(texts: list[str], batch_size: int = 32) -> list[dict]:
    """
    Run sentiment analysis on a list of social media texts.
    Returns list of {label, score} dicts.
    """
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        # Clean texts — remove URLs and excessive whitespace
        cleaned = [
            " ".join(word for word in t.split() if not word.startswith("http"))[:512]
            for t in batch
        ]
        preds = sentiment_pipeline(cleaned)
        for pred in preds:
            results.append({
                "sentiment": LABEL_MAP.get(pred["label"], pred["label"]),
                "confidence": round(pred["score"], 3),
            })
        print(f"  Processed {min(i + batch_size, len(texts))}/{len(texts)}")

    return results

def sentiment_report(df: pd.DataFrame, text_col: str = "text") -> pd.DataFrame:
    """
    Enrich a DataFrame with sentiment scores and produce a summary report.
    """
    texts = df[text_col].fillna("").tolist()
    print(f"Analysing sentiment for {len(texts)} posts...")
    sentiments = analyse_sentiment_batch(texts)

    df["sentiment"]   = [s["sentiment"]  for s in sentiments]
    df["confidence"]  = [s["confidence"] for s in sentiments]

    # Summary
    summary = df["sentiment"].value_counts(normalize=True).mul(100).round(1)
    print("\n── Sentiment Distribution ──")
    for label, pct in summary.items():
        bar = "█" * int(pct / 2)
        print(f"  {label:10s}: {bar} {pct}%")

    return df

# Apply to Reddit data
reddit_df = pd.read_csv("reddit_learnpython.csv")
reddit_df = sentiment_report(reddit_df, text_col="title")
reddit_df.to_csv("reddit_with_sentiment.csv", index=False)

# Apply to X data
tweets_df = pd.read_csv("tweets.csv")
tweets_df = sentiment_report(tweets_df, text_col="text")
tweets_df.to_csv("tweets_with_sentiment.csv", index=False)

Part 5: Building a Brand Monitoring Dashboard

Combine all sources into a unified brand monitoring pipeline:

import asyncio
import pandas as pd
from datetime import datetime, timezone

async def run_brand_monitor(brand_name: str, competitor: str = None) -> dict:
    """
    Collect brand mentions from Reddit and X, analyse sentiment,
    and produce a unified brand health report.
    """
    report = {
        "brand":       brand_name,
        "run_at":      datetime.now(timezone.utc).isoformat(),
        "reddit":      {},
        "twitter":     {},
        "sentiment":   {},
    }

    # ── Reddit ────────────────────────────────────────────────
    print(f"\n[1/3] Scraping Reddit for '{brand_name}'...")
    reddit_df = monitor_keyword_across_subreddits(
        keyword=brand_name,
        subreddits=["technology", "Python", "webdev", "datascience", "programming"],
        limit_per_sub=50
    )
    reddit_df = sentiment_report(reddit_df, text_col="title")

    report["reddit"] = {
        "total_posts":     len(reddit_df),
        "avg_score":       round(reddit_df["score"].mean(), 1),
        "total_comments":  int(reddit_df["comments"].sum()),
        "top_post":        reddit_df.sort_values("score", ascending=False).iloc[0]["title"],
        "top_subreddit":   reddit_df["subreddit"].value_counts().index[0],
    }

    # ── Sentiment ─────────────────────────────────────────────
    all_sentiments = pd.concat([
        reddit_df[["sentiment", "confidence"]],
    ])
    sentiment_counts = all_sentiments["sentiment"].value_counts(normalize=True).mul(100)
    report["sentiment"] = {
        "positive_pct": round(sentiment_counts.get("positive", 0), 1),
        "neutral_pct":  round(sentiment_counts.get("neutral", 0), 1),
        "negative_pct": round(sentiment_counts.get("negative", 0), 1),
        "overall":      sentiment_counts.idxmax(),
    }

    # ── Print Report ──────────────────────────────────────────
    print(f"\n{'═'*50}")
    print(f"  BRAND MONITOR: {brand_name.upper()}")
    print(f"  Run at: {report['run_at']}")
    print(f"{'═'*50}")
    print(f"\n  Reddit mentions:   {report['reddit']['total_posts']}")
    print(f"  Avg post score:    {report['reddit']['avg_score']}")
    print(f"  Most active sub:   r/{report['reddit']['top_subreddit']}")
    print(f"\n  Sentiment:")
    print(f"    ✅ Positive:  {report['sentiment']['positive_pct']}%")
    print(f"    😐 Neutral:   {report['sentiment']['neutral_pct']}%")
    print(f"    ❌ Negative:  {report['sentiment']['negative_pct']}%")
    print(f"    Overall:      {report['sentiment']['overall'].upper()}")
    print(f"\n  Top post: \"{report['reddit']['top_post'][:80]}...\"")

    return report

# Run
report = asyncio.run(run_brand_monitor("python scraping"))

Rate Limits and Platform Rules Reference

Platform	Daily free limit	Rate limit reset	Key restriction
X API v2 (Free)	500k tweets/month read	15 min windows	No historical data
X API v2 (Basic, $100/mo)	10M tweets/month	15 min windows	30-day history
Reddit PRAW	1,000 req/10 min	Rolling	Public posts only
Instagram (no auth)	~200 req/hour	Hourly	Public profiles only
Instagram (with session)	200–500 actions/day	Daily	Avoid aggressive scraping

FAQ

Q: snscrape is broken in 2026 — what should I use instead? snscrape's X scraper broke in 2023 when X removed the guest API and has not been reliably fixed. Use Tweepy with X API v2 for authorised projects. For DIY scraping, use the Playwright approach above.

Q: Can I scrape private Instagram posts? No — and you shouldn't. Scraping private account data without consent violates Instagram's Terms of Service, GDPR, and potentially criminal hacking laws in many jurisdictions. Only collect publicly available data.

Q: What's the best model for social media sentiment analysis in 2026?cardiffnlp/twitter-roberta-base-sentiment-latest is trained specifically on tweets and outperforms generic models. For Reddit, which is longer-form, distilbert-base-uncased-finetuned-sst-2-english works well. For multilingual content, use nlptown/bert-base-multilingual-uncased-sentiment.

Q: Reddit changed its API — is PRAW still free? Yes. PRAW's free tier (500 requests per 10 minutes) was not affected by the 2023 API changes. Only third-party apps accessing certain endpoints at very high volume were impacted. Standard research use via PRAW remains free.

Q: How do I handle deleted posts in my Reddit dataset? Posts with [deleted] author or [removed] body are common. Filter them before analysis: df = df[df['author'] != '[deleted]'] and df = df[df['text'] != '[removed]'].

Summary

Platform	Best Method	Key Library	What You Can Get
X / Twitter	Tweepy (API v2)	tweepy	Tweets, metrics, author data
X / Twitter	Playwright + session	playwright	Search results, profiles
Reddit	PRAW	praw	Posts, comments, subreddits
Instagram	httpx + cookies	httpx	Public profiles, post counts
All platforms	HuggingFace	transformers	Sentiment, emotion, topics

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Sign in With Google in Node.js — Without Passport

ZyVOP — Tue, 21 Jul 2026 06:14:19 +0000

Most "Sign in with Google" tutorials hand you Passport.js and a dozen middleware functions, then stop before explaining what any of them do. When the OAuth dance fails in production — wrong redirect URI, missing refresh token, expired state — you're left debugging a black box.

This post builds the whole flow from scratch: PKCE code generation, the authorization redirect, the callback handler that validates everything before touching the database, token exchange, and refresh token handling. No Passport, no auth library. Just four endpoints and the Google API.

This is the third post in Zyvop's auth series. If you haven't read the 2FA implementation or the magic link post yet, the session module here is the same one from those — a JWT in an httpOnly cookie. The new parts are everything that happens before you issue that cookie.

Source: https://github.com/zyvop27-cmyk/zyvop-blogs/tree/master/oauth-google

Setup

git clone [YOUR_GITHUB_REPO_URL]
cd oauth-google
npm install
cp .env.example .env

Then in Google Cloud Console:

APIs & Services → Credentials → Create Credentials → OAuth 2.0 Client ID
Application type: Web application
Authorized Redirect URIs: http://localhost:3000/auth/google/callback
Copy the Client ID and Secret into .env

Generate a JWT_SECRET:

node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"

npm start
# open http://localhost:3000

Why PKCE, and why does it matter for server-side apps

PKCE (Proof Key for Code Exchange) was designed for mobile and single-page apps that can't keep a client secret truly secret. For server-side apps you already have a secret, so adding PKCE is belt-and-suspenders. Google now requires it for all new OAuth clients regardless of app type, so it's no longer optional — it's just how OAuth with Google works in 2026.

The mechanism: before sending the user to Google, you generate a random code_verifier, hash it to produce a code_challenge, and send the challenge to Google.

When the user comes back with an authorization code, you exchange the code by also sending the original verifier. Google hashes it and checks it matches the challenge it stored. Anyone who intercepts the authorization code without the original verifier can't use it.

// src/lib/pkce.js
import { randomBytes, createHash } from "node:crypto";

export function generateVerifier() {
  // 32 bytes → 43-char URL-safe base64, no padding (RFC 7636)
  return randomBytes(32).toString("base64url");
}

export function deriveChallenge(verifier) {
  return createHash("sha256").update(verifier).digest("base64url");
}

export function verifyChallenge(verifier, expectedChallenge) {
  if (typeof verifier !== "string" || verifier.length < 43) return false;
  return deriveChallenge(verifier) === expectedChallenge;
}

base64url is the base64 alphabet with + → -, / → _, and all = padding stripped. Node has it built in since v14 — no library needed.

CSRF protection: the state parameter

The state parameter is how you prevent cross-site request forgery on the callback. You generate a random value, send it to Google, and Google sends it back in the callback. You verify it matches what you sent before touching anything else.

The state also carries the PKCE verifier. When the user lands on the callback, you need the verifier to complete the token exchange — but you can't put it in the URL, since that exposes it to server logs and referrer headers.

You could store it in a signed httpOnly cookie, but that means setting an additional cookie before the redirect and cleaning it up on callback. The state parameter is a neater single-point solution: one random value carries both the CSRF token and the verifier retrieval key.

// src/lib/state.js
import { randomBytes } from "node:crypto";

const _store = new Map();
const STATE_TTL_MS = 10 * 60 * 1000;

export function createState(verifier) {
  const state = randomBytes(24).toString("hex");
  _store.set(state, { verifier, expiresAt: Date.now() + STATE_TTL_MS });
  return state;
}

export function consumeState(state) {
  if (typeof state !== "string" || !state) return null;

  const entry = _store.get(state);
  _store.delete(state); // delete before returning — no replay window

  if (!entry) return null;
  if (Date.now() > entry.expiresAt) return null;
  return entry.verifier;
}

consumeState deletes the entry before checking it — not after. The order matters: if you check first and delete second, two simultaneous requests with the same state both pass the check before either deletes.

This way only one gets the verifier back; the other gets null.

The in-memory Map works fine for a single server process. For multiple instances behind a load balancer, move it to Redis with a 10-minute TTL.

Building the authorization URL

// src/lib/google.js
export function buildAuthUrl() {
  const verifier = generateVerifier();
  const challenge = deriveChallenge(verifier);
  const state = createState(verifier);

  const params = new URLSearchParams({
    client_id: process.env.GOOGLE_CLIENT_ID,
    redirect_uri: process.env.GOOGLE_REDIRECT_URI,
    response_type: "code",
    scope: "openid email profile",
    state,
    code_challenge: challenge,
    code_challenge_method: "S256",
    access_type: "offline",   // ask for a refresh token
    prompt: "consent",        // force consent screen so refresh token is always issued
  });

  return { url: `https://accounts.google.com/o/oauth2/v2/auth?${params}` };
}

access_type: "offline" tells Google you want a refresh token. Without it, you only get an access token that expires in an hour, and the user has to sign in again. prompt: "consent" forces the consent screen on every sign-in, which is the only reliable way to get a refresh token every time — Google silently skips it on returning users unless you force it.

The callback: three validations before anything else

When Google redirects the user back, the callback handler does three things before touching the database:

// src/routes/auth.js
router.get("/google/callback", async (req, res) => {
  const { code, state, error } = req.query;

  if (error) return res.redirect("/?error=access_denied");
  if (!code || !state) return res.redirect("/?error=invalid_callback");

  const verifier = consumeState(state);  // validates + consumes in one step
  if (!verifier) return res.redirect("/?error=invalid_state");

  try {
    const { accessToken, refreshToken } = await exchangeCode(code, verifier);
    const userInfo = await fetchUserInfo(accessToken);
    const user = upsertUser(db, userInfo, refreshToken);

    issueSession(res, { sub: user.google_sub, email: user.email,
                        name: user.name, picture: user.picture });
    res.redirect("/dashboard");
  } catch (err) {
    console.error("[oauth] callback failed:", err.message);
    res.redirect("/?error=auth_failed");
  }
});

Check error first — this is what Google sends when the user clicks "Cancel". Then check code and state are present.

consumeState in one call validates the state is real, not expired, and not already used, and returns the PKCE verifier. Only if all three pass do you make any network requests.

Token exchange and userinfo

The token exchange sends the authorization code plus the PKCE verifier to Google's token endpoint:

export async function exchangeCode(code, verifier, fetchFn = fetch) {
  const res = await fetchFn("https://oauth2.googleapis.com/token", {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      code,
      client_id: process.env.GOOGLE_CLIENT_ID,
      client_secret: process.env.GOOGLE_CLIENT_SECRET,
      redirect_uri: process.env.GOOGLE_REDIRECT_URI,
      grant_type: "authorization_code",
      code_verifier: verifier,
    }),
  });

  if (!res.ok) {
    const err = await res.text();
    throw new Error(`Token exchange failed (${res.status}): ${err}`);
  }
  const data = await res.json();
  return {
    accessToken: data.access_token,
    refreshToken: data.refresh_token ?? null,
    idToken: data.id_token,
    expiresIn: data.expires_in,
  };
}

Google's response includes an id_token — a JWT containing the user's identity claims. You could decode and verify it locally using Google's published JWK keys, which avoids a second network request.

This post uses the /userinfo endpoint instead, because it always returns current data and sidesteps implementing RS256 signature verification. For high-throughput production code, verify the id_token locally.

The userinfo call is straightforward:

export async function fetchUserInfo(accessToken, fetchFn = fetch) {
  const res = await fetchFn("https://www.googleapis.com/oauth2/v3/userinfo", {
    headers: { Authorization: `Bearer ${accessToken}` },
  });
  if (!res.ok) throw new Error(`Userinfo fetch failed (${res.status})`);

  const data = await res.json();
  if (!data.email_verified) throw new Error("Google account email is not verified");

  return { sub: data.sub, email: data.email, name: data.name ?? null,
           picture: data.picture ?? null, emailVerified: data.email_verified };
}

The email_verified check matters. Google allows accounts with unverified email addresses. Silently accepting one would let someone claim ownership of an email they don't control.

Storing users and handling refresh tokens

Users are keyed by Google's sub (subject), not email. Email addresses can change; sub is permanent for a given Google account. The upsert uses COALESCE to preserve the existing refresh token when the new one is null:

db.prepare(`
  INSERT INTO users (google_sub, email, name, picture, refresh_token)
  VALUES (@sub, @email, @name, @picture, @refreshToken)
  ON CONFLICT (google_sub) DO UPDATE SET
    email         = excluded.email,
    name          = excluded.name,
    picture       = excluded.picture,
    refresh_token = COALESCE(excluded.refresh_token, users.refresh_token),
    last_login    = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
  RETURNING *
`).get({ sub, email, name, picture, refreshToken: refreshToken ?? null });

Google only sends a refresh token on first sign-in (or after the user revokes access). If you overwrite the stored refresh token with null on subsequent logins, you lose the ability to refresh the user's access token without asking them to sign in again. COALESCE keeps the old one when the new one is absent.

When the access token expires and you need a new one:

export async function refreshAccessToken(refreshToken, fetchFn = fetch) {
  const res = await fetchFn("https://oauth2.googleapis.com/token", {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      client_id: process.env.GOOGLE_CLIENT_ID,
      client_secret: process.env.GOOGLE_CLIENT_SECRET,
      refresh_token: refreshToken,
      grant_type: "refresh_token",
    }),
  });
  if (!res.ok) {
    const err = await res.text();
    throw new Error(`Token refresh failed (${res.status}): ${err}`);
  }
  const data = await res.json();
  return { accessToken: data.access_token, expiresIn: data.expires_in };
}

A 400 response from this endpoint usually means the refresh token was revoked — the user removed your app from their Google account. Handle it by clearing the session and prompting a fresh sign-in.

The sign-in page

public/index.html is a single static page with a "Continue with Google" button that links to /auth/google. The interesting part is the error handling — when the OAuth flow fails for any reason, the server redirects back to /?error=<reason>. The page reads that query param and shows a human-readable message:

const ERRORS = {
  access_denied:    "You cancelled the sign-in. Try again when you're ready.",
  invalid_state:    "Something went wrong with the sign-in flow. Please try again.",
  invalid_callback: "The callback from Google was malformed. Please try again.",
  auth_failed:      "Sign-in failed. Please try again or contact support.",
  config:           "The server is misconfigured. Contact the site owner.",
  default:          "Something went wrong. Please try again."
};

const params = new URLSearchParams(location.search);
const errKey = params.get("error");
if (errKey) {
  const box = document.getElementById("error-box");
  box.textContent = ERRORS[errKey] ?? ERRORS.default;
  box.classList.add("visible");
}

Each error key maps to a specific failure point in the callback handler — access_denied when the user clicks Cancel, invalid_state when the state check fails, auth_failed when the token exchange or userinfo call throws. This matters: a generic "something went wrong" on a sign-in page sends users nowhere. Mapping errors to actionable messages costs five lines.

Testing without a Google account

Every external call takes an injectable fetchFn parameter. Tests pass a mock instead:

function mockFetch(status, body) {
  return async () => ({
    ok: status >= 200 && status < 300,
    status,
    json: async () => body,
    text: async () => JSON.stringify(body),
  });
}

test("exchangeCode throws on non-200 response", async () => {
  await assert.rejects(
    () => exchangeCode("code", "verifier", mockFetch(400, { error: "invalid_grant" })),
    /Token exchange failed/
  );
});

test("fetchUserInfo throws if email is not verified", async () => {
  await assert.rejects(
    () => fetchUserInfo("token", mockFetch(200, {
      sub: "123", email: "unverified@example.com", email_verified: false
    })),
    /email is not verified/
  );
});

The state store test exercises expiry by monkey-patching Date.now — no timers, no waiting:

test("expired state entries are not returned", () => {
  const realNow = Date.now;
  const state = createState("verifier");

  Date.now = () => realNow() + 11 * 60 * 1000; // 11 minutes later
  try {
    assert.equal(consumeState(state), null);
  } finally {
    Date.now = realNow;
  }
});

npm test
# tests 31 · pass 31 · fail 0

Trying it with curl

Once the server is running with real Google credentials:

Hit the sign-in page:

curl -I http://localhost:3000/auth/google

HTTP/1.1 302 Found
Location: https://accounts.google.com/o/oauth2/v2/auth?client_id=...&state=...&code_challenge=...

The OAuth flow goes through a browser — there's no way to automate it with curl end-to-end. To check the API after signing in, copy your session cookie from browser DevTools (Application → Cookies) and pass it directly:

curl -H "Cookie: session=<paste-your-jwt-here>" http://localhost:3000/api/me

{
  "sub": "1234567890",
  "email": "you@gmail.com",
  "name": "Your Name",
  "picture": "https://lh3.googleusercontent.com/..."
}

Refresh the access token (stored server-side):

curl -H "Cookie: session=<paste-your-jwt-here>" \
  -X POST http://localhost:3000/auth/refresh

{ "accessToken": "ya29.new-token", "expiresIn": 3599 }

No active session:

curl http://localhost:3000/api/me

{ "error": "Not authenticated." }

Before going to production

Move the state store to Redis — the in-memory Map disappears on restart and won't work across multiple server instances. The interface is the same, just backed by Redis keys with a 10-minute TTL.

Set NODE_ENV=production to enable the Secure flag on the session cookie, which browsers require over HTTPS. Add your production domain to Authorized Redirect URIs in Google Cloud Console — that list is an exact-match allowlist, and the callback will silently fail with redirect_uri_mismatch if your domain isn't on it.

If your threat model requires it, encrypt the refresh_token column at rest before writing to SQLite. A leaked database shouldn't hand an attacker long-lived access to user Google accounts.

Get the code: https://github.com/zyvop27-cmyk/zyvop-blogs/tree/master/oauth-google

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Build a URL Shortener With Click Analytics in Node.js

ZyVOP — Fri, 17 Jul 2026 11:18:46 +0000

Bit.ly and short.io are fine until you want to know which country your clicks are coming from, which referrer is driving traffic, and whether that campaign from last Tuesday is still converting. At that point, you're either paying for a premium plan or wishing you'd just built it yourself.

This post builds a complete URL shortener with analytics — short code generation, SQLite storage, IP geolocation, referrer tracking, and a dashboard showing daily click trends, browser breakdown, and country distribution.

No external analytics service, no database server to spin up. The whole thing runs on SQLite and ships as a single Node.js process.

Source code: https://github.com/zyvop27-cmyk/zyvop-blogs/tree/main/url-shortener

Stack and setup

Express 5 · SQLite (better-sqlite3) · geoip-lite · ua-parser-js · nanoid

git clone [YOUR_GITHUB_REPO_URL]
cd url-shortener
npm install
npm start

Open http://localhost:3000. The database creates itself in data/links.db on first run.

One environment variable is required before clicks will be tracked — HASH_SALT, used to hash IP addresses before storing them. Generate one:

node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"

Add it to .env. Without it, the server starts but logs a warning and skips click recording on every redirect.

Short code generation

The first decision: how to make the codes. Random is the right choice over sequential integers — sequential IDs are trivially enumerable, someone can increment from 1 and hit every link ever created.

Random codes with a good alphabet make that infeasible.

The alphabet deliberately excludes characters that look alike:

// src/lib/shortcode.js
import { customAlphabet } from "nanoid";

const ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjkmnpqrstuvwxyz23456789";
const CODE_LENGTH = 7;
const generate = customAlphabet(ALPHABET, CODE_LENGTH);

export function generateCode(db, existsFn, maxAttempts = 5) {
  for (let i = 0; i < maxAttempts; i++) {
    const code = generate();
    if (!existsFn(db, code)) return code;
  }
  throw new Error("Failed to generate a unique code after multiple attempts");
}

No 0, O, I, i, l, o, or 1 — seven characters that look like each other on printed paper or low-res screens. With 55 characters and 7 positions that's 55^7 ≈ 1.52 trillion possible codes. Collision handling is implemented correctly anyway (retry up to 5 times), but with that address space and any realistic link volume, it will never be needed.

The database

SQLite handles everything: URL storage, click records, and all the analytics queries. No Postgres, no Redis, no infrastructure to run. better-sqlite3 is synchronous, which means no connection pool to manage and no async/await ceremony around queries.

Two tables:

CREATE TABLE links (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  code       TEXT NOT NULL UNIQUE,
  url        TEXT NOT NULL,
  created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);

CREATE TABLE clicks (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  link_id    INTEGER NOT NULL REFERENCES links(id),
  clicked_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
  country    TEXT,
  referrer   TEXT,
  browser    TEXT,
  os         TEXT,
  ip_hash    TEXT
);

The analytics queries (top countries, referrers, daily trend) all run against the clicks table with a JOIN on links. The most complex one is the daily trend:

db.prepare(`
  SELECT
    strftime('%Y-%m-%d', clicked_at) AS date,
    COUNT(*) AS clicks
  FROM clicks WHERE link_id = ?
  GROUP BY date ORDER BY date DESC LIMIT 30
`).all(link.id);

SQLite's strftime handles the date bucketing natively. No date library needed.

One gotcha with the database module: better-sqlite3 is synchronous and stateful, so the module caches the connection after the first call. Tests need isolated in-memory databases, not the cached production one. The fix is simple — never cache :memory: connections:

export function getDb(dbPath = DB_PATH) {
  if (dbPath !== ":memory:" && _db) return _db;
  const db = new Database(dbPath);
  // ... schema setup ...
  if (dbPath !== ":memory:") _db = db;
  return db;
}

Each test gets its own fresh in-memory database. Nothing bleeds between test cases.

Recording clicks without slowing down redirects

The redirect handler records the click and then sends the 301. better-sqlite3 is synchronous — it blocks the Node.js event loop during the write — but WAL mode means it's appending to the write-ahead log file rather than syncing the main database.

A single INSERT completes in microseconds in practice. The try/catch ensures analytics failures never break a redirect:

// src/routes/redirect.js
const CODE_RE = /^[A-Za-z0-9]{4,10}$/;

router.get("/:code", (req, res) => {
  const { code } = req.params;
  if (!CODE_RE.test(code)) return res.status(404).send("Link not found.");

  const link = getLinkByCode(db, code);
  if (!link) return res.status(404).send("Link not found.");

  try {
    const clickData = parseClickData(req);
    insertClick(db, link.id, clickData);
  } catch (err) {
    console.error("[redirect] analytics error:", err.message);
  }

  res.redirect(301, link.url);
});

The /:code route must be registered last in server.js. It matches any single-segment path (including /healthz), so any named routes registered after it will never be reached. Express matches in registration order, not specificity order.

One other decision worth noting: 301 vs 302. A 301 (permanent redirect) tells browsers and search engines to cache the destination — follow the link once and the browser may never hit the shortener again.

That means if you later change the destination URL, existing users won't see the change. For a personal shortener that's usually fine; use 302 if you need updatable destinations.

The POST /api/shorten endpoint is rate-limited to 20 requests per hour per IP. Redirect and stats endpoints are not limited — a popular link shouldn't throttle its own clicks.

What gets tracked and how

Three pieces of data come in on every click: the IP address, the User-Agent header, and the Referer header.

// src/lib/analytics.js
export function parseClickData(req) {
  const ip = getClientIp(req);
  const ua = req.headers["user-agent"] ?? "";
  const rawReferrer = req.headers["referer"] || req.headers["referrer"] || null;

  const salt = process.env.HASH_SALT;
  if (!salt) throw new Error("HASH_SALT is not set. Add it to your .env file.");
  const ipHash = crypto.createHash("sha256").update(ip + salt).digest("hex").slice(0, 16);
  const geo = ip ? geoip.lookup(ip) : null;
  const country = geo?.country ?? null;

  const parsed = UAParser(ua);
  const browser = parsed.browser?.name ?? null;
  const os = parsed.os?.name ?? null;

  let referrer = null;
  if (rawReferrer) {
    try {
      referrer = new URL(rawReferrer).hostname;
    } catch {
      referrer = rawReferrer.slice(0, 100);
    }
  }

  return { country, referrer, browser, os, ipHash };
}

The IP gets hashed before storage. Raw IPs are PII — storing them without justification is a compliance headache in most jurisdictions.

A SHA-256 hash of the IP (keyed with a secret from your environment) lets you count unique visitors reliably without keeping the raw address. geoip-lite does the country lookup from a bundled local database, so there's no external API call.

Referrers get normalized to hostname only. https://google.com/search?q=example becomes google.com. The full URL isn't useful for the dashboard, and storing full search queries you didn't ask for is the kind of thing that causes problems later.

The dashboard

The dashboard is a single HTML file that calls the API. No framework:

GET /api/links — all shortened links with total click counts
GET /api/stats/:code — full breakdown for one link (by country, referrer, browser, daily trend)

The daily trend renders as a bar chart built from raw div elements. Bar height is a percentage of the maximum day's count, so no chart library dependency.

Clicking "Stats" on any row loads that link's analytics inline below the table. The stat panel shows total clicks, unique visitors (by hashed IP), and three bar charts. Country shows the top 10 countries, referrer shows where traffic came from, browser breaks down the client distribution.

Trying it with curl

Make sure HASH_SALT is set in .env before running these — without it the server starts and redirects work, but clicks won't be recorded and stats will show zero.

Shorten a URL:

curl -X POST http://localhost:3000/api/shorten \
  -H "Content-Type: application/json" \
  -d '{"url": "https://zyvop.com/building-a-production-ai-agent-in-node-js-tool-calling-the-react-loop-and-error-handling-zzftm}'

{
  "code": "P8WbGd7",
  "shortUrl": "http://localhost:3000/P8WbGd7",
  "url": "https://zyvop.com/building-a-production-ai-agent-in-nodejs"
}

Follow the redirect:

curl -I http://localhost:3000/P8WbGd7

HTTP/1.1 301 Moved Permanently
Location: https://zyvop.com/building-a-production-ai-agent-in-nodejs

Pull the stats:

curl http://localhost:3000/api/stats/P8WbGd7

{
  "link": { "code": "P8WbGd7", "url": "https://zyvop.com/...", "created_at": "2026-07-14T06:00:00Z" },
  "totalClicks": 3,
  "uniqueVisitors": 1,
  "byCountry": [{ "country": "US", "clicks": 3 }],
  "byReferrer": [{ "referrer": "google.com", "clicks": 2 }, { "referrer": "Direct", "clicks": 1 }],
  "byBrowser": [{ "browser": "Chrome", "clicks": 3 }],
  "dailyTrend": [{ "date": "2026-07-14", "clicks": 3 }]
}

"Direct" in byReferrer is what COALESCE(referrer, 'Direct') returns for clicks that arrived without a Referer header — typed directly into the browser bar, opened from a native app, or clicked from an email client that strips referrers.

Invalid URL:

curl -s -w " [%{http_code}]" -X POST http://localhost:3000/api/shorten \
  -H "Content-Type: application/json" \
  -d '{"url": "ftp://notallowed"}'

{"error":"url must be a valid http or https URL."} [400]

Unknown code:

curl -I http://localhost:3000/ZZZZZZZ

HTTP/1.1 404 Not Found

Tests

npm test

# tests 24
# pass  24
# fail   0

24 tests across three suites. The shortcode suite verifies the alphabet, the collision retry logic, and that it throws correctly when retries are exhausted.

The analytics suite tests IP hashing, country lookup, referrer normalization, and user-agent parsing — including the privacy property that raw IPs don't appear in the stored hash. The database suite runs every query against an in-memory SQLite instance: inserts, retrieval, unique visitor counting, daily trend grouping, and the unique constraint on codes.

Taking it further

The setup above works for personal use or a small team. A few things to add before opening it to the public:

Auth on the shorten endpoint — right now anyone who can reach the server can create links. A simple API key check is enough for most cases.
Custom slugs — let users specify POST /api/shorten with an optional customCode field; validate it against the same alphabet and check for conflicts before inserting.
Click buffering — if redirect volume gets high, the synchronous SQLite write on every click becomes the bottleneck. Buffer clicks in memory and flush in batches of 100 or every 5 seconds.
Link expiry — add an expires_at column to links and check it in the redirect handler.

Get the code: https://github.com/zyvop27-cmyk/zyvop-blogs/tree/main/url-shortener

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Four Ways a Refresh Token Request Fails — Only One Means Trouble

ZyVOP — Fri, 10 Jul 2026 07:35:46 +0000

A refresh token exists for one reason: exchange itself for a new access token, once, and then stop being useful. Everything about a good implementation follows from taking "once" literally. This one does — every refresh token is single-use, and grouped with every other token descended from the same login into a family. Calling POST /auth/refresh with a given token can fail four different ways, and three of them are just bookkeeping. The fourth is the one this post is actually about.

One row per token, one family per login

@Entity()
export class RefreshToken {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column()
  userId: string;

  @Index()
  @Column()
  familyId: string;

  @Index({ unique: true })
  @Column()
  tokenHash: string;

  @Column({ default: false })
  used: boolean;

  @Column({ default: false })
  revoked: boolean;

  @Column({ type: 'timestamp' })
  expiresAt: Date;

  @CreateDateColumn()
  createdAt: Date;
}

familyId is what makes "family" more than a metaphor — it's set once, at login, and then carried forward unchanged into every row that session's rotation produces afterward:

async login(@Body() dto: LoginDto) {
  const user = await this.authService.validatePassword(dto.email, dto.password);
  const accessToken = this.authService.issueAccessToken(user.id, user.email);
  const refresh = await this.refreshTokenService.issue(user.id); // no familyId = fresh session
  return { accessToken, refreshToken: refresh.rawToken };
}

Omitting familyId here is the signal that this is a brand new session rather than a rotation — issue() generates a fresh one. Every later call to issue() for this same chain, from inside rotate(), passes that same value back in instead. Two logins from the same user, at the same time, produce two completely independent families — there's nothing linking them beyond sharing a userId, which matters later.

Three ordinary rejections

A token that was never real. Someone sends a string that doesn't hash to anything in the table.

if (!stored) {
  throw new UnauthorizedException('Invalid refresh token');
}

curl -X POST http://localhost:3000/auth/refresh -H "Content-Type: application/json" \
  -d '{"refreshToken":"not-a-real-token-at-all"}'
# -> 401 {"message":"Invalid refresh token", ...}

A token that aged out. Refresh tokens in this implementation live 7 days by default. Past that, they're just gone.

if (stored.expiresAt < new Date()) {
  throw new UnauthorizedException('Refresh token has expired');
}

Confirmed directly rather than waiting a week: backdate a token's expiresAt in Postgres by a day, then try to use it —

{"message":"Refresh token has expired","error":"Unauthorized","statusCode":401}

— exactly the branch that's supposed to fire.

A token whose session already ended. Logout, or (as covered below) a reuse event elsewhere in the same family, flips a revoked flag.

if (stored.revoked) {
  throw new UnauthorizedException('Refresh token has been revoked');
}

None of these three are interesting on their own — they're the normal outcomes of a token being wrong, old, or deliberately ended. The fourth rejection reason looks identical from the outside (same 401, same shape) but means something completely different underneath.

The fourth: a token that's already been spent

if (stored.used) {
  await this.revokeFamily(stored.familyId);
  throw new UnauthorizedException('Refresh token reuse detected; session revoked');
}

stored.used = true;
await this.refreshTokenRepository.save(stored);

return this.issue(stored.userId, stored.familyId);

Every successful refresh marks the token it consumed as used before issuing its replacement. A legitimate client only ever moves forward through that chain — it gets a new token and uses that one next time, never the old one again.

So if an already-used token shows up in a request, the client presenting it isn't following the chain. Either it's a genuine client retrying a request whose response got lost on the way back (a real possibility, not a hypothetical), or it's a second party holding a copy of a token the real owner has already moved past.

There's no way to tell those apart from inside this one request — both look exactly like "this exact token, again" — so both get treated as the same signal: something's wrong with this session, not just this token.

That's why the response isn't "reject this token and move on." It's revokeFamily, which doesn't touch just the token that got reused — it kills every unrevoked row sharing that familyId, including whichever token the legitimate client is currently holding as its "real" one.

Proving that actually happens, not just reading the code and assuming: rotate once (RT1 → RT2), then try RT1 again.

curl -X POST http://localhost:3000/auth/refresh -H "Content-Type: application/json" -d '{"refreshToken":"<RT1>"}'
# -> 401 {"message":"Refresh token reuse detected; session revoked", ...}

RT1 failing is expected — it's already used. The real test is what happens to RT2, which was never itself reused, was legitimately issued, and — if the family weren't revoked — should still work fine:

curl -X POST http://localhost:3000/auth/refresh -H "Content-Type: application/json" -d '{"refreshToken":"<RT2>"}'
# -> 401 {"message":"Refresh token has been revoked", ...}

RT2 dies too, and the error message even confirms why — not "reuse detected" (that already happened, on RT1) but plain "revoked," which is exactly what a downstream consequence of someone else's reuse should look like from RT2's own perspective. This is the actual security property, not the headline description of it: a compromise anywhere in the chain takes down the entire chain, including the part that was never touched.

The blast radius runs the other direction too, and it's worth confirming it actually stops where it should: a second user, bob@example.com, with his own completely separate family, refreshes without incident while all of the above is happening to alice's account. Nothing about bob's session is touched — revokeFamily only ever operates on rows sharing the one familyId it was given, and bob's family was never that one. The failure is total within a compromised family and entirely absent outside it.

Logging out is the same mechanism, aimed on purpose

Every revocation above happened as a side effect of something going wrong. Logout is the identical underlying operation — revokeFamily — called directly, when nothing is wrong at all:

async revokeFamilyByToken(rawToken: string): Promise<void> {
  const stored = await this.refreshTokenRepository.findOne({
    where: { tokenHash: this.hash(rawToken) },
  });
  if (stored) {
    await this.revokeFamily(stored.familyId);
  }
}

curl -X POST http://localhost:3000/auth/logout -H "Content-Type: application/json" \
  -d '{"refreshToken":"<RT>"}'
# -> { "message": "Logged out" }

curl -X POST http://localhost:3000/auth/refresh -H "Content-Type: application/json" \
  -d '{"refreshToken":"<RT>"}'
# -> 401 {"message":"Refresh token has been revoked", ...}

One question this raises immediately: does logging out on one device end every session, or just the one that called it? Worth an actual answer instead of a guess — log in twice for the same user (call them a phone session and a laptop session, two independent families), log out using only the phone's token, then try refreshing each:

phone session, after logout: {"message":"Refresh token has been revoked", ...}
laptop session, untouched:   { "accessToken": "...", "refreshToken": "..." }   (succeeds normally)

Logging out ends the one family attached to the token that was actually presented. It has no way to reach a different family for the same user, because nothing about revokeFamily looks at userId at all — only familyId. Two logins are two unrelated chains that happen to belong to the same person, and ending one says nothing about the other.

What doesn't die with the family

Everything above is about the refresh token. The access token issued alongside RT2 — call it AT2 — is a separate, stateless JWT with its own 15-minute clock, and revoking a refresh family doesn't reach into that clock:

curl http://localhost:3000/auth/me -H "Authorization: Bearer <AT2>"
# -> 200 { "id": "...", "email": "alice@example.com" }

Issued before the reuse event above, and it still works after the whole family got revoked. That's not a gap in the implementation — it's the actual reason access tokens are kept short-lived in the first place. There's no database row for an access token to flip a revoked bit on; the only thing bounding how long a leaked one stays useful is its own expiry. Fifteen minutes is the real ceiling on "how bad is it if this specific token leaks," independent of anything the refresh layer detects or reacts to.

Why the token hash is SHA-256, not bcrypt

The password column in this same project uses bcrypt, and it should — passwords are short, human-chosen, and drawn from a relatively small effective space, so a hash that's deliberately slow is what makes brute-forcing a stolen hash impractical.

A refresh token is a different kind of secret: 40 bytes of crypto.randomBytes, 320 bits of pure randomness, with no human-guessable structure to exploit. Brute-forcing a hash of that is infeasible no matter how fast the hash function is, so bcrypt's deliberate slowness buys nothing here — it just adds cost to every single refresh call for no security benefit. A plain, fast sha256 is the correct tool once the secret's strength comes from entropy rather than from making guessing expensive.

Trying this out, and what it doesn't cover

npm install && cp .env.example .env
docker run --name refresh-postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=refresh_demo -p 5432:5432 -d postgres:16
npm run start:dev

synchronize: true handles table creation for this demo; swap it for real migrations before this touches production. Register and log in to get real values for the <RT1>/<RT2>/<AT2> placeholders used throughout above:

curl -X POST http://localhost:3000/auth/register -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"correct-horse-battery"}'

curl -X POST http://localhost:3000/auth/login -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"correct-horse-battery"}'
# -> { "accessToken": "<AT1>", "refreshToken": "<RT1>" }

Left out on purpose:

A cleanup job for expired and revoked rows. They'll otherwise just accumulate — the BullMQ patterns from elsewhere in this series are a natural fit for sweeping them out.
Rate limiting on /auth/login and /auth/refresh. Both are obvious targets, and the earlier rate-limiting post in this series covers them directly.
A grace period for the "legitimate retry" half of the reuse ambiguity. Some production systems let a just-rotated token keep working for a few seconds, specifically to absorb a lost-response retry before treating reuse as a hard signal. This implementation takes the strict stance instead — simpler to reason about, simpler to verify — which is a deliberate tradeoff, not an oversight.

The full project, including the parts of this walkthrough not reproduced above, is in the refresh-token-rotation-demo repository next to this post.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Brick: The LLM Router That Skips the Cascade and Still Cuts Your Bill

ZyVOP — Wed, 08 Jul 2026 05:53:41 +0000

Most teams solve "which model should answer this prompt" one of two ways.

Pick one strong model and pay frontier prices for every request, including the ones a 9B open-weight model could handle blind. Or build a cascade: try the cheap model first, escalate on a low-confidence signal, and pay for every miss in tokens and latency.

Brick, from Italian inference provider Regolo.ai, is a third option. It reads a prompt once, scores it against a pool of models across six capability dimensions, and picks a single backend in one shot: no retries, no escalation ladder. Just model: "brick" dropped into an OpenAI-compatible call.

It's a young project with a real, published paper behind it: single digits on GitHub stars as of this writing, one CLI release tag, and a couple of distribution channels still marked "pending" in its own README. Here's what's actually going on under the hood, plus a benchmark nuance worth understanding before you cite the headline number.

What Brick Actually Routes To

Brick sits in front of a pool of models you define, then dispatches each incoming query to whichever one is the cheapest that can still get the answer right. The project frames three situations where that's worth doing:

You already run several models and want each query landing on the right one instead of a manual pick.
You're paying for Claude Code or Codex at frontier rates and want easy turns to land on a cheaper model automatically.
You want one endpoint in front of a mixed pool of OpenAI, GLM, DeepSeek, Kimi, and Qwen models instead of wiring each one up separately.

The default shipped config points every backend at Regolo's own API, but the pool is swappable. Anything that speaks the OpenAI chat-completions format can join.

How the Routing Decision Actually Gets Made

This is the part worth reading the source for, because the README summary undersells how mechanical it is. Every text query goes through two classifiers before a single token gets forwarded anywhere.

Capability vector. A ModernBERT classifier scores the query across six dimensions: coding, creative synthesis, instruction following, math reasoning, planning/agentic behavior, and world knowledge. The output is a soft probability distribution over those six, not a hard label.

Complexity score. A separate Qwen3.5-0.8B model with a LoRA adapter buckets the query into easy, medium, or hard. In the shipped config this runs as a remote call against Regolo's brick-complexity-pro endpoint rather than a local model.

The objective. Every model in the pool also has a fixed six-dimensional skill vector plus a cost weight. Brick computes J_m = D_m + β·a_m for each candidate, where D_m is the distance between the query's capability vector and that model's skill vector, and a_m is a normalized cost penalty.

It picks the model with the smallest J_m. That's the whole decision: one geometric comparison, argmin over the pool.

A separate r knob, ranging from -1 to 1, slides the whole pool between these two poles. Toward -1 favors the cheapest capable model; toward 1 favors the strongest one.

The paper's routing math behind that knob has more moving parts than a single weight, with separate calibration branches for the cost and quality directions. At the API level, though, it's just one number you set at deploy time. The router's own config file, pulled straight from the repo, shows what the pool actually looks like:

models:
  - model: "qwen3.5-9b"
    skill_vector: [0.714788, 0.511538, 0.810109, 0.912146, 0.577072, 0.179876]
    use_reasoning: false
    cost_weight: 0.10
  - model: "deepseek-v4-flash"
    skill_vector: [0.820939, 0.657845, 0.863112, 0.934963, 0.62055, 0.488518]
    use_reasoning: false
    cost_weight: 0.40
  - model: "kimi2.6"
    skill_vector: [0.904272, 0.751595, 0.87018, 0.943892, 0.641863, 0.344074]
    use_reasoning: true
    reasoning_effort: "medium"
    cost_weight: 0.60

Two more blocks sit on top of the math for manual overrides, both documented in apps/router/README.md. keyword_rules in override mode hard-forces specific queries to a named model regardless of what the classifiers say.

The shipped default sends anything containing "debug," "refactor," or "write a function" straight to kimi2.6. A bias mode nudges one capability dimension up without forcing a decision, useful for tilting language-specific queries toward the coding axis.

Multimodal input skips the whole pipeline. Images and audio get preprocessed through OCR or Whisper-compatible speech-to-text first, then either routed as extracted text or forwarded straight to a vision model.

Claude Code and Codex Get a Dedicated Integration

This is where Brick is clearly aiming most of its early adoption. Running brick claude on wires an ANTHROPIC_BASE_URL override into ~/.claude/settings.json and starts a local router.

A new brick-claude option shows up in Claude Code's /model picker, sitting next to the built-in opus/sonnet/haiku aliases rather than replacing them. Five modes control the cost/quality trade-off, and they map directly onto Claude Code's existing thinking-effort slider:

Effort slider	Brick mode	Behavior
low	eco	always haiku
medium	lite	graduated tier, cheaper end
high	mid (default)	graduated tier, balanced
xhigh	pro	graduated tier, stronger end
max	max	always opus

The README confirms the two end tiers precisely: eco always picks haiku, max always picks opus. It doesn't spell out the exact easy/medium/hard-to-model mapping for the three tiers in between, only that they sit on a cost/quality gradient.

Selecting a native model name in the picker bypasses Brick entirely and forwards the request unchanged. That gives you an easy escape hatch if the router ever picks wrong.

A brick claude status command opens a live dashboard showing routed-by-model counts, per-model reasoning-effort distribution, the classifier's easy/medium/hard mix, and an estimated savings percentage against an all-opus baseline.

Codex support exists too, gated as beta, sharing the same five modes and status view. The two share host port 8000, though, so only one can serve at a time.

The Numbers, and What They're Actually Measuring

The project has a real published paper behind it: Brick: Spatial Capability Routing for the Mixture-of-Models (MoM) Paradigm, by Francesco Massa (Regolo.ai) and Marco Cristofanilli (Seeweb, where Regolo.ai is built), posted to arXiv in June 2026.

It reports results on a 5,504-query benchmark called Dataset A, spanning six capability dimensions. Each query is graded by protocol-specific checkers, an LLM judge, or both depending on the task type.

Setting	Accuracy	Cost per call	Latency
Always Qwen3.5-9b	63.17%	$0.0014 (1.0×)	8.1s*
Always DeepSeek-v4-flash	73.69%	$0.0029 (2.1×)	14.7s*
Always Kimi2.6	75.02%	$0.0307 (22.15×)	51.2s
Brick, min-cost profile (r=-1)	63.17%	$0.0014 (1.0×)	9.4s*
Brick, neutral profile (r=0)	74.11%	≈$0.0065 (4.7×)	—
Brick, max-quality profile (r=1)	76.98%	unconfirmed†	22.8s
Oracle bound (3-model pool)	83.25%	n/a	n/a

*Accuracy and per-call dollar cost are from the paper's own cost audit, priced off OpenRouter listings at evaluation time.

Latency figures marked with an asterisk are from the project's README summary table rather than the paper itself. The paper's own headline latency claim: median end-to-end latency drops from 51.2s (always-Kimi) to 22.8s (Brick at max-quality).

†The max-quality profile's cost multiple sits further down the same results table in the paper, past the section I could extract. Treat that cell as unverified rather than a claim either way.

Note also that the project's README reports its own, different, more rounded cost multiples for these same rows (1.0× / 4.0× / 6.0× / 1.5×). This table uses the paper's precise per-call dollar figures instead, since they're traceable to an actual pricing snapshot rather than a rounded illustration.

Against external baselines: Brick's low-cost profile beats FrugalGPT's 69.42% by 2.2 points at comparable cost, and its neutral profile beats Cascade Routing's 73.40% by 0.71 points.

That's the deployment-facing number: did the model Brick picked actually answer correctly. The paper is explicit that this isn't the only question worth asking.

It also defines a second metric, route-exact accuracy: does the router's pick match the literal cheapest model that could have solved the query. Queries none of the three models can solve still count against every router here, oracle included.

That makes it a stricter, cost-efficiency-facing bar than the headline number. Credit where it's due: the paper reports both metrics side by side in its main results table, not buried somewhere else.

On that stricter metric, a trivial "always pick the cheapest model" baseline scores 63.17% by definition, since the cheapest model is the correct pick on 63% of the queries. The interesting comparison is the other routers against that same bar.

Always-Kimi and RouteLLM both dispatch nearly everything to Kimi, and score only 21.28% and 21.31% route-exact accuracy despite matching Kimi's headline number. That's the paper's own evidence that neither is doing real query-level judgment.

FrugalGPT and Cascade Routing land at 31.03% and 28.96%. Brick clears 40.35% at a neutral setting and 63.17% at its cheapest profile, beating every external baseline on the metric that's supposed to be hardest on it.

One loose end worth flagging for anyone who goes digging in the repo themselves: a separate, Italian-language working file in packages/evals/baselines/RESULTS.md reports a route-exact figure for Brick of 46.37%. That number doesn't line up exactly with any of the r-settings in the paper's published table (63.17% / 48.66% / 40.35% at min/low/neutral).

That file reads like an earlier or differently-configured pipeline run than what shipped in the final paper, not a contradiction of it. If you're citing Brick's numbers for something that matters, pull them from the arXiv table rather than that file.

Trying It

The fastest path is the CLI, which self-hosts the router and wires it into Claude Code in two commands. It needs Node 18+ and Docker:

git clone https://github.com/regolo-ai/brick-SR1.git
cd brick-SR1/apps/cli && npm install && npm run build && npm link

brick claude on

For a raw OpenAI-compatible gateway without the CLI, the project's Docker image is documented but, as of this writing, still pending its first push to GHCR under the v2.1.0 tag. Once published, the documented usage looks like this:

docker run --rm -p 18000:18000 \
  -e REGOLO_API_KEY=$REGOLO_API_KEY \
  ghcr.io/regolo-ai/brick:latest

curl http://localhost:18000/v1/chat/completions \
  -H "Authorization: Bearer $REGOLO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"brick","messages":[{"role":"user","content":"Prove that sqrt(2) is irrational"}]}'

The response carries an x-selected-model header telling you which backend actually answered. A math-proof prompt should land on a reasoning-capable model; a bare "hello" should land on the cheapest one in the pool.

I'm quoting these commands straight from the project's own quickstart rather than claiming to have run them myself. With the Docker image and npm package both still pending publication, and no Regolo API key available in a sandboxed environment, there was nothing live to hit.

Where This Project Actually Sits Right Now

A few things worth knowing before you build on this. The router itself descends from vllm-project/spatial-router, an Apache-2.0 project.

Regolo's team added the six-dimension capability classifier, the complexity-score integration, the skill-distance objective, multimodal preprocessing, and the Claude Code passthrough on top. That lineage is disclosed plainly in the repo's NOTICE file, which is exactly how attribution should work on a fork.

As of this writing the repo has zero forks and a GitHub star count still in the single digits. That's normal for a project a few weeks old and not itself a signal of quality either way.

The codebase is a genuine multi-language build. Go and Rust make up the router (Rust handles ML embeddings via candle and classical ML via Linfa, compiled to shared libraries and linked through CGO), Python runs the training and eval pipelines, and TypeScript powers the CLI.

License is Apache-2.0 throughout. Whether it earns a permanent spot in your stack comes down to which problem you actually have.

If you're routing across a real multi-model pool, the paper's numbers are worth your own re-run before you trust them at production scale. If you just want Claude Code to stop burning Opus on "fix this typo" requests, the Claude-specific integration is the more immediately useful half of the project — and it works whether or not you buy the Dataset A accuracy claims at all.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Python Scraping at Scale: Distributed Crawling Across Multiple Machines (2026)

ZyVOP — Tue, 07 Jul 2026 06:04:28 +0000

Introduction: When One Machine Is No Longer Enough

Most Python developers write simple scrapers—requests, BeautifulSoup, a loop, CSV writer—just to get data once or twice. When scaling or running them over months, the real challenge is building a robust, scheduled system, not the parsing logic. The extraction code is tiny; the critical components are the queue, cache, storage, block detection, recovery loop, and exporters that keep the scraper alive against real‑world internet conditions.

At small scale, a single async Python process can handle thousands of pages per hour. But when you need to crawl millions of pages per day — competitor catalogues, job markets, news archives, e-commerce databases — a single machine hits hard limits: one IP address, one CPU, one point of failure.

Distributed scraping involves spreading tasks across multiple machines to increase speed and volume. Throttling and introducing random delays between requests can help to prevent IP bans, while rotating proxies help distribute requests and avoid detection. Managing sessions and leveraging parallel processing can further enhance efficiency.

This guide shows you exactly how to build a distributed scraping system that runs across multiple machines — using scrapy-redis for shared request queues, Docker for containerisation, and a production proxy management layer that survives real-world conditions.

The Architecture: One Queue, Many Workers

The core insight behind distributed scraping is simple: replace Scrapy's in-memory request queue with a shared Redis queue that every worker machine can read from.

┌─────────────────────────────────────────────────────────────┐
│                    DISTRIBUTED SCRAPER                      │
│                                                             │
│  Master                                                     │
│  ┌──────────┐    Seeds requests    ┌─────────┐             │
│  │  Spider  │──────────────────────▶│  Redis  │             │
│  │ (seed)   │                      │  Queue  │             │
│  └──────────┘                      └────┬────┘             │
│                                         │                   │
│  Workers (any number of machines)       │                   │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐                  │
│  │ Worker 1 │  │ Worker 2 │  │ Worker N │  ← pull jobs     │
│  │ (spider) │  │ (spider) │  │ (spider) │                  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘                  │
│       │              │              │                        │
│       └──────────────┴──────────────┘                       │
│                          │                                   │
│                    ┌─────▼──────┐                           │
│                    │  MongoDB   │  ← all workers write here │
│                    │ (results)  │                           │
│                    └────────────┘                           │
└─────────────────────────────────────────────────────────────┘

scrapy-redis lets multiple spider instances across machines pull from the same request queue with deduplication and job scheduling. This is the standard approach for large-scale web scraping that Python teams use in production.

Part 1: Setting Up scrapy-redis

pip install scrapy scrapy-redis redis pymongo

The distributed spider

# spiders/distributed_product_spider.py
import scrapy
from scrapy_redis.spiders import RedisSpider
from urllib.parse import urljoin
from datetime import datetime, timezone

class DistributedProductSpider(RedisSpider):
    """
    A Scrapy spider that pulls start URLs from a Redis list
    instead of a hardcoded start_urls list.

    To start crawling, push seed URLs to Redis:
        redis-cli lpush products:start_urls "https://example-store.com/products/"

    Any number of workers running this spider will cooperatively
    process the shared queue — each URL is processed exactly once.
    """

    name         = "distributed_products"
    redis_key    = "products:start_urls"   # Redis list to pop URLs from

    # How many URLs to pop from Redis at once per worker
    redis_batch_size = 16

    # Spider-level settings — override settings.py per spider
    custom_settings = {
        "CONCURRENT_REQUESTS":            32,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 8,
        "DOWNLOAD_DELAY":                 0.5,
        "RANDOMIZE_DOWNLOAD_DELAY":       True,
        "AUTOTHROTTLE_ENABLED":           True,
        "AUTOTHROTTLE_TARGET_CONCURRENCY": 16.0,
        "AUTOTHROTTLE_MAX_DELAY":         5.0,
        "RETRY_TIMES":                    3,
        "RETRY_HTTP_CODES":              [429, 500, 502, 503, 504],
    }

    def parse(self, response):
        """
        Parse a product listing page.
        Yields product detail requests AND discovers pagination links.
        """
        # Follow product links to detail pages
        for href in response.css("a.product-link::attr(href)").getall():
            yield response.follow(href, callback=self.parse_product)

        # Auto-discover pagination — push next page back to Redis queue
        next_page = response.css("a[rel='next']::attr(href)").get()
        if next_page:
            # Use follow() to handle relative URLs
            yield response.follow(next_page, callback=self.parse)

    def parse_product(self, response):
        """Extract product data from a detail page."""
        yield {
            "url":        response.url,
            "title":      response.css("h1::text").get("").strip(),
            "price":      response.css(".price::text").get("").strip(),
            "sku":        response.css("[data-sku]::attr(data-sku)").get(),
            "in_stock":   bool(response.css(".in-stock")),
            "description":response.css(".product-description::text").get("").strip()[:500],
            "scraped_at": datetime.now(timezone.utc).isoformat(),
            "worker_id":  self.settings.get("WORKER_ID", "unknown"),
        }

settings.py for distributed mode

# settings.py

BOT_NAME = "distributed_scraper"
SPIDER_MODULES = ["spiders"]

# ── scrapy-redis settings ─────────────────────────────────────
SCHEDULER            = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS     = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL            = "redis://redis-host:6379"

# Keep crawl state across restarts — resumable crawls
SCHEDULER_PERSIST    = True

# How long to wait for new URLs before worker shuts down
SCHEDULER_IDLE_BEFORE_CLOSE = 30

# ── MongoDB pipeline ──────────────────────────────────────────
ITEM_PIPELINES = {
    "pipelines.MongoPipeline":        100,
    "pipelines.DuplicateFilterPipeline": 50,
}
MONGO_URI      = "mongodb://mongo-host:27017/"
MONGO_DATABASE = "distributed_scrape"

# ── Concurrency ───────────────────────────────────────────────
CONCURRENT_REQUESTS              = 32
CONCURRENT_REQUESTS_PER_DOMAIN   = 8
DOWNLOAD_DELAY                   = 0.5
RANDOMIZE_DOWNLOAD_DELAY         = True

# ── Retry ─────────────────────────────────────────────────────
RETRY_ENABLED    = True
RETRY_TIMES      = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504, 522, 524]

# ── Logging ───────────────────────────────────────────────────
LOG_LEVEL = "INFO"

# ── Downloader middlewares ────────────────────────────────────
DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": None,
    "middlewares.RotatingProxyMiddleware": 350,
    "middlewares.UserAgentRotationMiddleware": 400,
    "scrapy.downloadermiddlewares.retry.RetryMiddleware": 550,
}

# ── User agent pool ───────────────────────────────────────────
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/119.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64) Chrome/118.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
]

Part 2: Production Middlewares

# middlewares.py
import random
import logging
from scrapy import signals
from scrapy.exceptions import NotConfigured

logger = logging.getLogger(__name__)

class UserAgentRotationMiddleware:
    """Rotate User-Agent on every request."""

    def __init__(self, user_agents):
        self.user_agents = user_agents

    @classmethod
    def from_crawler(cls, crawler):
        agents = crawler.settings.getlist("USER_AGENTS")
        if not agents:
            raise NotConfigured("USER_AGENTS not set")
        return cls(agents)

    def process_request(self, request, spider):
        request.headers["User-Agent"] = random.choice(self.user_agents)

class RotatingProxyMiddleware:
    """
    Rotate through a proxy pool.
    Tracks failures per proxy and removes bad proxies from the pool.
    """

    def __init__(self, proxies, max_failures=5):
        self.proxies      = list(proxies)
        self.failures     = {}
        self.max_failures = max_failures

    @classmethod
    def from_crawler(cls, crawler):
        proxies = crawler.settings.getlist("PROXY_LIST", [])
        if not proxies:
            raise NotConfigured("PROXY_LIST is empty")
        return cls(proxies)

    def process_request(self, request, spider):
        if not self.proxies:
            return  # No proxies left — run without
        proxy = random.choice(self.proxies)
        request.meta["proxy"] = proxy

    def process_response(self, request, response, spider):
        if response.status in (403, 407, 429):
            proxy = request.meta.get("proxy")
            self._mark_failure(proxy)
        return response

    def process_exception(self, request, exception, spider):
        proxy = request.meta.get("proxy")
        self._mark_failure(proxy)

    def _mark_failure(self, proxy):
        if not proxy:
            return
        self.failures[proxy] = self.failures.get(proxy, 0) + 1
        if self.failures[proxy] >= self.max_failures:
            if proxy in self.proxies:
                self.proxies.remove(proxy)
                logger.warning(f"Removed bad proxy: {proxy} ({self.max_failures} failures)")

class BlockDetectionMiddleware:
    """
    Detect common block patterns and trigger retries with a different proxy.
    """

    BLOCK_PATTERNS = [
        "access denied", "captcha", "blocked", "forbidden",
        "unusual traffic", "robot", "automated queries",
        "cf-browser-verification", "ddos-guard",
    ]

    def process_response(self, request, response, spider):
        body_lower = response.text[:2000].lower()

        is_blocked = (
            response.status in (403, 429, 503) or
            any(p in body_lower for p in self.BLOCK_PATTERNS) or
            len(response.text) < 300
        )

        if is_blocked:
            logger.warning(f"Block detected on {request.url} — retrying")
            request.meta["proxy"]    = None  # Force new proxy on retry
            request.dont_filter      = True
            return request           # Re-schedule the request

        return response

Part 3: MongoDB Pipeline with Bulk Writes

# pipelines.py
import logging
from datetime import datetime, timezone
from pymongo import MongoClient, UpdateOne
from pymongo.errors import BulkWriteError
from itemadapter import ItemAdapter

logger = logging.getLogger(__name__)

class DuplicateFilterPipeline:
    """Track seen URLs in memory to drop duplicates before DB write."""

    def open_spider(self, spider):
        self.seen_urls = set()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        url = adapter.get("url", "")
        if url in self.seen_urls:
            from scrapy.exceptions import DropItem
            raise DropItem(f"Duplicate URL: {url}")
        self.seen_urls.add(url)
        return item

class MongoPipeline:
    """
    Write scraped items to MongoDB using bulk operations.
    Upserts on URL — safe to re-run without creating duplicates.
    """

    BULK_SIZE  = 200   # Flush every 200 items
    COLLECTION = "products"

    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db  = mongo_db
        self._buffer   = []

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get("MONGO_URI", "mongodb://localhost:27017/"),
            mongo_db =crawler.settings.get("MONGO_DATABASE", "scrapy_data"),
        )

    def open_spider(self, spider):
        self.client     = MongoClient(self.mongo_uri)
        self.col        = self.client[self.mongo_db][self.COLLECTION]
        self.col.create_index("url", unique=True)
        self.items_written = 0
        logger.info(f"MongoDB connected: {self.mongo_db}.{self.COLLECTION}")

    def close_spider(self, spider):
        if self._buffer:
            self._flush()
        self.client.close()
        logger.info(f"MongoDB closed. Total items written: {self.items_written}")

    def process_item(self, item, spider):
        self._buffer.append(dict(item))
        if len(self._buffer) >= self.BULK_SIZE:
            self._flush()
        return item

    def _flush(self):
        ops = [
            UpdateOne({"url": doc["url"]}, {"$set": doc}, upsert=True)
            for doc in self._buffer
        ]
        try:
            result = self.col.bulk_write(ops, ordered=False)
            count  = result.upserted_count + result.modified_count
            self.items_written += count
            logger.info(f"Flushed {len(self._buffer)} items → MongoDB")
        except BulkWriteError as e:
            logger.error(f"Bulk write error: {e.details.get('writeErrors', [])[:2]}")
        finally:
            self._buffer.clear()

Part 4: Docker Compose — Multi-Worker Setup

# docker-compose.yml
version: "3.9"

x-worker-base: &worker-base
  build: .
  volumes:
    - .:/app
  environment:
    - REDIS_URL=redis://redis:6379
    - MONGO_URI=mongodb://mongo:27017/
    - PROXY_LIST=${PROXY_LIST}
  depends_on:
    - redis
    - mongo
  restart: unless-stopped

services:

  # ── Infrastructure ──────────────────────────────────────────
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
    volumes: ["redis_data:/data"]

  mongo:
    image: mongo:7
    ports: ["27017:27017"]
    volumes: ["mongo_data:/data/db"]

  # ── Scrapy Workers ──────────────────────────────────────────
  # Run as many of these as you have cores / IPs
  worker-1:
    <<: *worker-base
    command: >
      scrapy crawl distributed_products
      -s WORKER_ID=worker-1
      -s CONCURRENT_REQUESTS=16
    environment:
      - REDIS_URL=redis://redis:6379
      - MONGO_URI=mongodb://mongo:27017/
      - WORKER_ID=worker-1

  worker-2:
    <<: *worker-base
    command: >
      scrapy crawl distributed_products
      -s WORKER_ID=worker-2
      -s CONCURRENT_REQUESTS=16
    environment:
      - REDIS_URL=redis://redis:6379
      - MONGO_URI=mongodb://mongo:27017/
      - WORKER_ID=worker-2

  worker-3:
    <<: *worker-base
    command: >
      scrapy crawl distributed_products
      -s WORKER_ID=worker-3
      -s CONCURRENT_REQUESTS=16
    environment:
      - REDIS_URL=redis://redis:6379
      - MONGO_URI=mongodb://mongo:27017/
      - WORKER_ID=worker-3

  # ── Seed Service — pushes start URLs into Redis ─────────────
  seeder:
    <<: *worker-base
    command: python seeder.py
    restart: "no"   # Run once then exit

volumes:
  redis_data:
  mongo_data:

Part 5: The Seeder — Feeding URLs Into the Queue

# seeder.py
import redis
import time
import sys
from urllib.parse import urlencode

REDIS_URL   = "redis://localhost:6379"
REDIS_KEY   = "products:start_urls"

def seed_from_list(urls: list[str], batch_size: int = 500):
    """Push a list of start URLs into the Redis queue."""
    r = redis.from_url(REDIS_URL)

    # Clear existing queue if resuming fresh
    existing = r.llen(REDIS_KEY)
    if existing > 0:
        print(f"Queue already has {existing:,} URLs. Adding to it.")

    pushed = 0
    for i in range(0, len(urls), batch_size):
        batch = urls[i:i + batch_size]
        r.rpush(REDIS_KEY, *batch)
        pushed += len(batch)
        print(f"Seeded {pushed:,}/{len(urls):,} URLs")

    print(f"\nDone. Redis queue '{REDIS_KEY}' has {r.llen(REDIS_KEY):,} URLs.")

def seed_paginated_site(
    base_url: str,
    start_page: int = 1,
    end_page: int = 500,
    page_param: str = "page"
):
    """Generate paginated URLs and push to Redis."""
    urls = []
    for page in range(start_page, end_page + 1):
        params = {page_param: page}
        urls.append(f"{base_url}?{urlencode(params)}")

    seed_from_list(urls)

def monitor_queue():
    """Monitor queue depth and worker progress in real time."""
    r = redis.from_url(REDIS_URL)
    print("Monitoring queue depth (Ctrl+C to stop)...")
    try:
        while True:
            depth   = r.llen(REDIS_KEY)
            seen    = r.scard(f"{REDIS_KEY}:dupefilter") or 0
            print(f"  Queue: {depth:,} pending | Seen: {seen:,} processed", end="\r")
            time.sleep(2)
    except KeyboardInterrupt:
        print("\nMonitor stopped.")

if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "seed"

    if mode == "seed":
        seed_paginated_site(
            base_url  = "https://example-store.com/products",
            start_page = 1,
            end_page   = 1000,
        )
    elif mode == "monitor":
        monitor_queue()
    elif mode == "clear":
        r = redis.from_url(REDIS_URL)
        r.delete(REDIS_KEY)
        print(f"Queue '{REDIS_KEY}' cleared.")

Part 6: Scaling to Kubernetes

For truly large-scale crawling across dozens of machines, Kubernetes is the standard deployment target. The key insight: each Scrapy worker is a stateless pod that reads from the shared Redis queue.

# k8s/scrapy-worker-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: scrapy-workers
  labels:
    app: scrapy-worker
spec:
  replicas: 10   # Start with 10 workers; scale up/down with kubectl
  selector:
    matchLabels:
      app: scrapy-worker
  template:
    metadata:
      labels:
        app: scrapy-worker
    spec:
      containers:
        - name: scrapy-worker
          image: your-registry/scrapy-worker:latest
          command:
            - scrapy
            - crawl
            - distributed_products
          env:
            - name: REDIS_URL
              valueFrom:
                secretKeyRef:
                  name: scraper-secrets
                  key: redis-url
            - name: MONGO_URI
              valueFrom:
                secretKeyRef:
                  name: scraper-secrets
                  key: mongo-uri
            - name: WORKER_ID
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name  # Pod name as worker ID
          resources:
            requests:
              memory: "256Mi"
              cpu:    "250m"
            limits:
              memory: "512Mi"
              cpu:    "500m"

Scale workers up or down instantly:

# Scale to 20 workers
kubectl scale deployment scrapy-workers --replicas=20

# Check worker status
kubectl get pods -l app=scrapy-worker

# View logs from all workers
kubectl logs -l app=scrapy-worker --tail=50

# Auto-scale based on Redis queue depth (requires custom metrics)
kubectl autoscale deployment scrapy-workers --min=2 --max=50

Part 7: Proxy Management at Scale

For serious scraping in 2026, residential proxies are almost always the safer option. Proxy quality matters far more than proxy quantity. A smaller pool of clean residential IPs usually performs much better than massive low-quality networks.

Here's a production proxy manager with health checking:

# proxy_manager.py
import asyncio
import httpx
import random
import time
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ProxyHealth:
    url:              str
    success_count:    int   = 0
    failure_count:    int   = 0
    last_used:        float = 0.0
    last_success:     float = 0.0
    avg_response_ms:  float = 0.0
    is_banned:        bool  = False

    @property
    def success_rate(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0.0

    @property
    def score(self) -> float:
        """Composite score: higher = better proxy to use."""
        if self.is_banned:
            return 0.0
        recency_bonus = max(0, 1 - (time.time() - self.last_success) / 3600)
        speed_score   = max(0, 1 - self.avg_response_ms / 5000)
        return self.success_rate * 0.6 + recency_bonus * 0.2 + speed_score * 0.2

class ProxyPool:
    """
    Intelligent proxy pool with health tracking and weighted selection.
    Workers register success/failure, pool learns which proxies are best.
    """

    BAN_THRESHOLD = 0.2   # Mark as banned if success rate drops below 20%

    def __init__(self, proxy_urls: list[str]):
        self.proxies = {url: ProxyHealth(url=url) for url in proxy_urls}

    def get_proxy(self, strategy: str = "weighted") -> Optional[str]:
        """Select a proxy using the specified strategy."""
        available = [
            p for p in self.proxies.values()
            if not p.is_banned
        ]

        if not available:
            return None

        if strategy == "random":
            return random.choice(available).url

        elif strategy == "weighted":
            # Weight by score — best proxies get used more often
            scores = [max(p.score, 0.01) for p in available]
            total  = sum(scores)
            weights = [s / total for s in scores]
            return random.choices(available, weights=weights)[0].url

        elif strategy == "round_robin":
            # Sort by last_used timestamp — use least recently used
            available.sort(key=lambda p: p.last_used)
            return available[0].url

        return available[0].url

    def report_success(self, proxy_url: str, response_ms: float):
        if proxy_url in self.proxies:
            p = self.proxies[proxy_url]
            p.success_count  += 1
            p.last_used       = time.time()
            p.last_success    = time.time()
            # Rolling average response time
            p.avg_response_ms = (p.avg_response_ms * 0.8 + response_ms * 0.2)

    def report_failure(self, proxy_url: str):
        if proxy_url in self.proxies:
            p = self.proxies[proxy_url]
            p.failure_count += 1
            p.last_used      = time.time()
            # Auto-ban proxies with very low success rate
            if p.failure_count > 10 and p.success_rate < self.BAN_THRESHOLD:
                p.is_banned = True
                print(f"  Proxy banned (success rate {p.success_rate:.0%}): {proxy_url}")

    def get_stats(self) -> dict:
        active  = [p for p in self.proxies.values() if not p.is_banned]
        banned  = [p for p in self.proxies.values() if p.is_banned]
        avg_sr  = sum(p.success_rate for p in active) / len(active) if active else 0

        return {
            "total":    len(self.proxies),
            "active":   len(active),
            "banned":   len(banned),
            "avg_success_rate": f"{avg_sr:.1%}",
            "best_proxy": max(active, key=lambda p: p.score).url if active else None,
        }

async def health_check_proxies(pool: ProxyPool, test_url: str = "https://httpbin.org/ip"):
    """
    Periodically check all proxies and un-ban those that have recovered.
    Run this as a background task.
    """
    async with httpx.AsyncClient(timeout=10) as client:
        for url, proxy in list(pool.proxies.items()):
            if not proxy.is_banned:
                continue
            try:
                start = time.time()
                r = await client.get(test_url, proxies={"https": url})
                if r.status_code == 200:
                    elapsed_ms    = (time.time() - start) * 1000
                    proxy.is_banned = False
                    pool.report_success(url, elapsed_ms)
                    print(f"  Proxy recovered: {url}")
            except Exception:
                pass   # Still banned

    stats = pool.get_stats()
    print(f"Proxy pool: {stats['active']} active, {stats['banned']} banned, "
          f"avg success rate: {stats['avg_success_rate']}")

Part 8: Monitoring Your Distributed Crawl

# monitor.py — real-time crawl progress dashboard
import redis
import time
import json
from pymongo import MongoClient
from datetime import datetime

def live_dashboard(redis_url: str, mongo_uri: str, refresh_seconds: int = 5):
    """Print a live crawl progress dashboard to the terminal."""
    r   = redis.from_url(redis_url)
    db  = MongoClient(mongo_uri)["distributed_scrape"]

    start_time   = time.time()
    prev_count   = 0

    try:
        while True:
            # Queue stats
            queue_depth  = r.llen("products:start_urls")
            seen_count   = r.scard("products:start_urls:dupefilter") or 0

            # DB stats
            items_stored = db["products"].count_documents({})
            items_delta  = items_stored - prev_count
            rate_per_min = items_delta * (60 / refresh_seconds)
            prev_count   = items_stored

            # Worker stats (scrapy-redis stores worker heartbeats)
            workers = r.smembers("scrapy:workers") or set()

            # Elapsed
            elapsed = time.time() - start_time
            h, m    = divmod(int(elapsed), 3600)
            m, s    = divmod(m, 60)

            print(f"\033[2J\033[H")   # Clear screen
            print(f"{'═'*55}")
            print(f"  DISTRIBUTED SCRAPE MONITOR — {datetime.now().strftime('%H:%M:%S')}")
            print(f"{'═'*55}")
            print(f"  Runtime:       {h:02d}h {m:02d}m {s:02d}s")
            print(f"  Active workers:{len(workers)}")
            print(f"{'─'*55}")
            print(f"  Queue depth:   {queue_depth:>10,}  (URLs remaining)")
            print(f"  URLs seen:     {seen_count:>10,}  (deduplicated total)")
            print(f"  Items stored:  {items_stored:>10,}  (in MongoDB)")
            print(f"  Rate:          {rate_per_min:>10.0f}  items/minute")
            print(f"{'─'*55}")

            if rate_per_min > 0 and queue_depth > 0:
                eta_mins = queue_depth / rate_per_min
                h2, m2   = divmod(int(eta_mins * 60), 3600)
                m2, s2   = divmod(m2, 60)
                print(f"  ETA:           {h2:02d}h {m2:02d}m  (estimated)")

            print(f"{'═'*55}")
            time.sleep(refresh_seconds)

    except KeyboardInterrupt:
        print("\nMonitor stopped.")

if __name__ == "__main__":
    live_dashboard(
        redis_url="redis://localhost:6379",
        mongo_uri="mongodb://localhost:27017/",
    )

Part 9: Resumable Crawls

One of the biggest advantages of scrapy-redis is that crawls are inherently resumable. If a worker crashes or you need to add more workers mid-crawl, simply restart:

# Start a fresh crawl
python seeder.py seed

# Launch workers (they'll pick up from where they left off if SCHEDULER_PERSIST=True)
docker-compose up --scale worker=5

# Pause all workers (Ctrl+C in docker-compose)
# Resume later — queue state is preserved in Redis
docker-compose up --scale worker=10   # Can add more workers

To completely reset a crawl:

# Clear the queue and deduplication filter
redis-cli del products:start_urls
redis-cli del products:start_urls:dupefilter
python seeder.py seed   # Re-seed with fresh URLs

Performance Numbers: What to Expect

One machine first, then scale out. Tune Scrapy performance optimization settings before throwing hardware at the problem — raise CONCURRENT_REQUESTS, turn on AUTOTHROTTLE, and enable HTTP caching. If one server can't handle scraping millions of pages, set up a shared queue for distributed crawling across multiple workers.

Real-world throughput benchmarks:

Setup	Pages/hour	Cost estimate
1 worker, no proxy	~8,000	Free
1 worker + 10 proxies	~25,000	~$5/day
5 workers + 50 proxies	~120,000	~$20/day
20 workers + 200 proxies	~500,000	~$80/day
100 workers (Kubernetes)	~2,500,000	~$350/day

Summary

Component	Tool	Role
Spider	Scrapy + scrapy-redis	Crawl logic + distributed request handling
Queue	Redis	Shared URL queue with built-in deduplication
Worker deployment	Docker Compose / Kubernetes	Horizontal scale, stateless workers
Proxy management	Custom ProxyPool	Health-tracked, weighted proxy selection
Storage	MongoDB (bulk upserts)	Centralised, deduplicated results
Monitoring	Custom dashboard + Flower	Real-time progress and worker health
Resumability	SCHEDULER_PERSIST=True	Crash-safe, restartable crawls

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

AI Hype Meets Reality: Security Risks, Thin ROI, and the Rise of Skepticism | The AI Daily Roundup

ZyVOP — Mon, 06 Jul 2026 03:00:08 +0000

Trend Overview: From Hype to Hard Evidence

Across today’s headlines a single narrative emerges – the AI boom is colliding with hard‑won lessons about security, economics, and user agency. Companies are pulling back on unvetted models, researchers are exposing the limits of productivity gains, and regulators and users alike are demanding concrete safeguards.

Why the Shift Matters

When AI tools promise “instant expertise” or “autonomous vulnerability discovery,” the cost of failure is no longer abstract. A data breach, a wasted budget, or a legal liability can cripple a firm faster than a missed hype cycle. Simultaneously, the promised “AI‑powered productivity” is proving to be a marginal gain that evaporates before it reaches a paycheck. The convergence of security scares and thin ROI forces senior leaders to re‑evaluate AI adoption strategies, shifting capital toward proven, auditable solutions.

Evidence from the Day’s Stories

Security Backlash

Alibaba bans Claude Code after internal alarms about a potential backdoor. The move signals that even tech giants will enforce “zero‑trust” policies when model provenance is uncertain.
Claude Mythos preview triggers a 3.5× spike in high‑severity CVEs. Anthropic’s own showcase of autonomous vulnerability discovery has unintentionally amplified the attack surface, prompting a wave of disclosures from Microsoft, Google, Apple, and AWS partners.
Claude’s “memory” feature under fire. Engineers report zero performance benefit from retaining session transcripts, highlighting a broader pattern: added complexity without measurable security or efficiency gains.

Economic Reality Check

Study finds AI saves only ~3% of work hours, and less than 5% of that time translates into higher pay. The gap between lab‑controlled speedups (15‑55%) and real‑world payroll data underscores a “leaky bucket” problem.
Yann LeCun admits current AI isn’t “smart” and bets on next‑generation systems that can handle real‑world data. His candid assessment validates the productivity findings: today’s models excel at narrow tasks but falter on general, embodied intelligence.

User Control & Transparency

Kagi adds an AI toggle, letting users disable AI‑driven search features. The option reflects growing demand for “opt‑out” mechanisms when AI adds noise rather than value.
Google’s Gemini Code Assist is being retired after less than a year, suggesting that even well‑funded services struggle to sustain adoption when the perceived ROI is low.

Who Wins, Who Loses

Beneficiaries: Security‑focused vendors (vulnerability‑management platforms, zero‑trust providers), audit‑ready AI platforms that expose provenance, and enterprises that adopt a measured, task‑specific AI strategy.

Losers: Companies banking on blanket AI adoption without clear use‑case validation, hype‑driven product launches, and consultants who sell “AI transformation” without quantifiable outcomes.

What Changes Next?

Stricter governance: Expect more corporate bans similar to Alibaba’s, and internal policies that require independent model audits before deployment.
Metrics‑first adoption: Teams will benchmark AI impact against payroll and P&L data before scaling, mirroring the methodology of the Danish study.
Feature pruning: Products that add “bells and whistles” (e.g., session‑transcript memory, generic code assistants) will be trimmed or sunset unless they demonstrate clear ROI.
Regulatory focus on AI‑driven security: As vulnerability spikes become visible, regulators may mandate disclosure of AI‑generated exploits and require companies to certify model safety.

Bottom Line

The AI industry is entering a phase of “skeptical scaling.” Security incidents, modest productivity gains, and user‑driven demand for control are forcing a recalibration. Leaders who embed rigorous measurement, enforce zero‑trust model policies, and prioritize real‑world value will capture the next wave of AI‑enabled advantage.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Passwordless Login With Magic Links in Node.js

ZyVOP — Sun, 05 Jul 2026 03:28:14 +0000

I've seen developers implement magic link auth four different ways and get it wrong three of them. The broken versions all share the same flaw: they use a regular get + del to verify tokens, which means two simultaneous requests with the same token both pass.

User A clicks the link in Gmail. The prefetch scanner in their email client hits it half a second earlier. Server issues two sessions. Neither user knows, but one of them is now authenticated in a context they didn't control.

This post builds it right. By the end you'll have a working Node.js implementation with proper atomic token verification, email enumeration protection, an Ethereal fallback so you can see the full flow without any SMTP configuration, and 19 tests that run without Redis or a real mail server.

Source code: http://github.com/zyvop27-cmyk/zyvop-blogs/tree/main/magic-link-auth

The one decision that matters most

Before touching any code: the difference between a safe magic link implementation and a broken one is a single Redis command.

The broken pattern:

const email = await redis.get(key);
if (email) await redis.del(key);
return email;

Between the get and the del, another request can read the same key. Both get a valid email back, both get sessions issued.

getDel collapses that into one atomic operation:

const email = await redis.getDel(key);
return email ?? null;

Redis executes this as a single command. No window — one request gets the email, any concurrent request gets null.

That's why Redis fits this problem better than a Postgres row. You'd need a transaction and an advisory lock to get the same guarantee from a relational database.

The full verify function adds a length check before the Redis call. A token that's too short or too long gets rejected immediately, no round-trip needed:

// src/lib/token.js
import crypto from "node:crypto";

const TOKEN_PREFIX = "magic:";
const TOKEN_TTL_SECONDS = 15 * 60;
const TOKEN_BYTES = 32;

export async function createToken(redis, email) {
  const token = crypto.randomBytes(TOKEN_BYTES).toString("hex");
  await redis.set(TOKEN_PREFIX + token, email.toLowerCase(), { EX: TOKEN_TTL_SECONDS });
  return token;
}

export async function verifyToken(redis, token) {
  if (!token || typeof token !== "string" || token.length !== TOKEN_BYTES * 2) {
    return null;
  }
  return (await redis.getDel(TOKEN_PREFIX + token)) ?? null;
}

crypto.randomBytes(32) gives 256 bits of entropy — 64 hex characters. Brute-forcing that against a 15-minute window isn't happening. The email gets lowercased on creation so User@Example.COM and user@example.com don't end up as separate keys that never match.

Redis also handles expiry natively via the EX option. No cron job to clean up stale tokens, no WHERE expires_at < NOW() queries — they disappear on their own.

What the user actually sees

The sign-in page is minimal HTML with one real piece of JavaScript: it handles the ?error=link_invalid query param that the server redirects to when a token has expired or already been used.

<!-- public/index.html (abbreviated) -->
<form id="form">
  <input type="email" id="email" placeholder="you@example.com" required>
  <button type="submit">Send sign-in link</button>
</form>
<div id="msg" class="message"></div>

<script>
  const ERRORS = {
    link_invalid: "That link has expired or already been used. Please request a new one.",
    default: "Something went wrong. Please try again."
  };

  // Show error from redirect after a failed verify attempt
  const errKey = new URLSearchParams(location.search).get("error");
  if (errKey) showMsg(ERRORS[errKey] ?? ERRORS.default, "error");

  document.getElementById("form").addEventListener("submit", async (e) => {
    e.preventDefault();
    const email = document.getElementById("email").value.trim();
    const btn = e.target.querySelector("button");

    btn.disabled = true;
    btn.textContent = "Sending…";

    const res = await fetch("/auth/request", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ email })
    });

    const data = await res.json();

    if (res.ok) {
      showMsg("Check your inbox — a sign-in link is on its way.", "success");
      document.getElementById("form").style.display = "none";
    } else {
      showMsg(data.error ?? ERRORS.default, "error");
      btn.disabled = false;
      btn.textContent = "Send sign-in link";
    }
  });
</script>

The ?error=link_invalid flow matters more than it looks. When verification fails — expired token, already used, someone typed a URL wrong — it redirects to /?error=link_invalid instead of returning a JSON 400.

Users arrive at /auth/verify by clicking a link in their email client, not via a fetch call. A JSON error on a blank page is a dead end; a redirect back to the sign-in form with a clear message is something a person can act on.

Sending the email without configuring SMTP

Production email uses SMTP_HOST, SMTP_PORT, SMTP_USER, and SMTP_PASS. Without those, the mailer silently creates an Ethereal test account and logs a preview URL. Ethereal is a real catch-all SMTP service — emails don't deliver anywhere, but you can open the URL and see the full rendered email, including the magic link button, right in your browser.

// src/lib/mailer.js
import nodemailer from "nodemailer";

let transporter = null;

async function getTransporter() {
  if (transporter) return transporter;

  if (process.env.SMTP_HOST) {
    transporter = nodemailer.createTransport({
      host: process.env.SMTP_HOST,
      port: Number(process.env.SMTP_PORT) || 587,
      secure: process.env.SMTP_SECURE === "true",
      auth: { user: process.env.SMTP_USER, pass: process.env.SMTP_PASS }
    });
  } else {
    const testAccount = await nodemailer.createTestAccount();
    transporter = nodemailer.createTransport({
      host: "smtp.ethereal.email",
      port: 587,
      auth: { user: testAccount.user, pass: testAccount.pass }
    });
    console.log(`[mailer] Ethereal test account: ${testAccount.user}`);
  }

  return transporter;
}

export async function sendMagicLink(to, magicUrl) {
  const transport = await getTransporter();

  const info = await transport.sendMail({
    from: process.env.EMAIL_FROM || '"Magic Link Auth" <no-reply@example.com>',
    to,
    subject: "Your sign-in link",
    text: `Sign-in link (expires in 15 minutes, single use):\n\n${magicUrl}`,
    html: `
      <p>Click below to sign in. Expires in <strong>15 minutes</strong>, single use.</p>
      <p style="margin:24px 0">
        <a href="${magicUrl}" style="background:#111;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none">
          Sign in
        </a>
      </p>
      <p style="color:#999;font-size:12px">Didn't request this? Safe to ignore.</p>
    `
  });

  if (!process.env.SMTP_HOST) {
    console.log(`[mailer] Preview URL: ${nodemailer.getTestMessageUrl(info)}`);
  }
}

The transporter is cached in a module-level variable. nodemailer.createTestAccount() makes a real HTTP call to Ethereal — you don't want that per-request.

For production, Resend and Postmark are the cleaner options over raw SMTP. They handle deliverability, bounce handling, and SPF/DKIM automatically. Hook them in via the SMTP_HOST and friends in .env.

Sessions, cookies, and why SameSite: lax is correct here

After verification, the token is gone and the user needs something that persists across requests. A JWT in an httpOnly cookie is the right shape: stateless (no server-side session table), survives page refreshes, inaccessible to JavaScript running on the page.

// src/lib/session.js
import jwt from "jsonwebtoken";

const COOKIE_NAME = "session";
const SESSION_TTL_SECONDS = 7 * 24 * 60 * 60;

function getSecret() {
  const secret = process.env.JWT_SECRET;
  if (!secret || secret.length < 32) {
    throw new Error("JWT_SECRET must be set and at least 32 characters long");
  }
  return secret;
}

export function issueSession(res, email) {
  const token = jwt.sign({ email }, getSecret(), { expiresIn: SESSION_TTL_SECONDS });
  res.cookie(COOKIE_NAME, token, {
    httpOnly: true,
    secure: process.env.NODE_ENV === "production",
    sameSite: "lax",
    maxAge: SESSION_TTL_SECONDS * 1000
  });
}

export function requireAuth(req, res, next) {
  const token = req.cookies?.[COOKIE_NAME];
  if (!token) return res.status(401).json({ error: "Not authenticated." });

  try {
    const payload = jwt.verify(token, getSecret());
    req.user = { email: payload.email };
    next();
  } catch {
    res.clearCookie(COOKIE_NAME);
    return res.status(401).json({ error: "Session expired. Please sign in again." });
  }
}

sameSite: "lax" needs explaining here because it interacts directly with how magic links work. When a user clicks a link in their email client, that's a top-level navigation — the browser follows it like a normal page load. Lax allows the cookie to be sent on those top-level navigations from external origins.

Strict would block that, requiring another sign-in if a user arrives from any external link. None requires HTTPS everywhere and opens CSRF exposure. Lax is the right call for this pattern.

secure only goes on in production. Without that conditional, local development over HTTP would silently fail to set the cookie and you'd spend an hour wondering why sessions don't persist.

The two routes that do the work

// src/routes/auth.js
import { Router } from "express";
import { createToken, verifyToken } from "../lib/token.js";
import { sendMagicLink as defaultSendMagicLink } from "../lib/mailer.js";
import { issueSession } from "../lib/session.js";

const EMAIL_RE = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;

export function createAuthRouter(redis, { sendMagicLink = defaultSendMagicLink } = {}) {
  const router = new Router();

  router.post("/request", async (req, res) => {
    const email = (req.body?.email ?? "").trim().toLowerCase();

    if (!EMAIL_RE.test(email)) {
      return res.status(400).json({ error: "A valid email address is required." });
    }

    try {
      const token = await createToken(redis, email);
      const appUrl = process.env.APP_URL || `http://localhost:${process.env.PORT || 3000}`;
      await sendMagicLink(email, `${appUrl}/auth/verify?token=${token}`);
      res.json({ ok: true, message: "Check your inbox for a sign-in link." });
    } catch (err) {
      console.error("[auth] request failed:", err);
      res.status(500).json({ error: "Failed to send the sign-in link. Please try again." });
    }
  });

  router.get("/verify", async (req, res) => {
    const email = await verifyToken(redis, req.query.token);
    if (!email) return res.redirect("/?error=link_invalid");
    issueSession(res, email);
    res.redirect("/dashboard");
  });

  router.post("/logout", (req, res) => {
    res.clearCookie("session");
    res.redirect("/");
  });

  return router;
}

/request returns 200 whether the email exists in your system or not. A 404 for unknown emails would tell anyone who tries that an address isn't registered — a user enumeration leak. The response is always "check your inbox," whether you're a real user or someone probing your database.

The sendMagicLink function is passed in as a default parameter, not imported at the top. That's the dependency injection hook tests use — swap it for a no-op, no SMTP connection required.

Running the whole thing

# Start Redis (Docker is fastest)
docker run -d -p 6379:6379 redis:alpine

# Install deps and start
npm install && cp .env.example .env && npm start

The server needs at minimum a JWT_SECRET in .env. Generate one:

node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"

Request a link:

curl -X POST http://localhost:3000/auth/request \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com"}'

{ "ok": true, "message": "Check your inbox for a sign-in link." }

Your server console will print something like:

[mailer] Preview URL: https://ethereal.email/message/WaQKMgKddxQDoou

Open that URL, click the "Sign in" button in the rendered email. You'll land on the dashboard. The token is gone — clicking the link again redirects to /?error=link_invalid.

Check your session after clicking (saves cookie to cookies.txt):

curl -c cookies.txt -b cookies.txt http://localhost:3000/api/me

{ "email": "you@example.com" }

Invalid email format:

curl -s -X POST http://localhost:3000/auth/request \
  -H "Content-Type: application/json" \
  -d '{"email": "notanemail"}'

{ "error": "A valid email address is required." }

Rate limiter — 5 requests per 15 minutes per IP, sixth gets a 429:

for i in $(seq 1 6); do
  curl -s -o /dev/null -w "request $i -> %{http_code}\n" \
    -X POST http://localhost:3000/auth/request \
    -H "Content-Type: application/json" \
    -d '{"email":"test@example.com"}'
done

request 1 -> 200
request 2 -> 200
request 3 -> 200
request 4 -> 200
request 5 -> 200
request 6 -> 429

Five is tight enough to stop inbox flooding and loose enough that a real user who typo'd their address gets a few retries.

Expired or already-used token:

curl -v "http://localhost:3000/auth/verify?token=$( python3 -c 'print("a"*64)')" 2>&1 | grep "Location:"

< Location: /?error=link_invalid

Tests

19 tests, no Redis or SMTP connection needed. The token suite runs against a plain in-memory Map mimicking the Redis interface; the session suite sets JWT_SECRET in before() and cleans up after; the route suite injects the no-op mailer:

npm test

# tests 19
# pass  19
# fail   0

If you want to verify the atomic single-use behavior beyond the unit test, run the server with a real Redis instance and hit /auth/verify with the same token from two curl commands fired in parallel:

TOKEN="paste-a-real-token-from-console-here"
curl "http://localhost:3000/auth/verify?token=$TOKEN" &
curl "http://localhost:3000/auth/verify?token=$TOKEN" &
wait

One will redirect to /dashboard. The other will redirect to /?error=link_invalid. That's getDel doing its job.

Before going live

Set NODE_ENV=production — the Secure cookie flag only activates in production, and without HTTPS the cookie won't be sent by browsers at all. Point REDIS_URL at a managed instance (Upstash has a free tier and works well with this setup). Swap Ethereal for a transactional email provider; Resend has a generous free tier and a clean Node.js SDK. Put the whole thing behind nginx or Caddy for TLS termination.

One thing the repo doesn't include but you'll want eventually: a users table or equivalent. Right now any email can request a link and get a session — there's no concept of "registered users." Adding a check in /request that verifies the email exists in your database before sending the link is one line, but the shape of that check depends on your stack.

Get the code: http://github.com/zyvop27-cmyk/zyvop-blogs/tree/main/magic-link-auth

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Rate Limiting Alone Won't Stop a Patient Attacker

ZyVOP — Sat, 04 Jul 2026 02:29:26 +0000

@nestjs/throttler counts requests per IP address within a time window. That's the entire mechanism — no concept of accounts, passwords, or "this one person is being targeted." Point it at a login endpoint and it will correctly stop a script hammering /login a hundred times a second. It will do nothing against someone guessing a password once every few minutes, or spreading attempts across a dozen IPs, because that was never the problem it's built to solve.

Below is both halves of a real defense: the IP throttle for volume, and a small Redis-backed service that locks by email for everything the throttle misses. Every response shape, status code, and timing claim here came from actually running the requests, not from the docs — including one interaction between two timers that will quietly undo a lockout's own "fresh start" if you're not paying attention to how they relate.

The throttle, and what it actually returns

A baseline policy on every route, plus a stricter override on login specifically:

// app.module.ts — global baseline: 20 requests/60s per IP, everywhere
ThrottlerModule.forRoot([{ name: 'default', ttl: 60000, limit: 20 }]),
providers: [{ provide: APP_GUARD, useClass: ThrottlerGuard }],

// auth.controller.ts — login gets a stricter limit than the rest of the API
@Throttle({ default: { limit: 10, ttl: 60000 } })
@Post('login')
async login(@Body() dto: LoginDto) { /* ... */ }

Whether that override replaces the global 20/60s or stacks on top of it isn't obvious from reading the decorator, so it's worth confirming rather than guessing: it replaces. A route-level @Throttle with the same policy name (default) fully takes over for that route. Eleven rapid requests to /auth/register (no override, rides the global policy) all came back 201; the same eleven against /auth/login would have tripped its 10-request limit well before the last one.

The response headers on a request that's still under the limit:

X-RateLimit-Limit: 10
X-RateLimit-Remaining: 9
X-RateLimit-Reset: 60

And once it's crossed:

// HTTP 429
{ "statusCode": 429, "message": "ThrottlerException: Too Many Requests" }

That's the library's real, unmodified default — not a placeholder for something cleaner. It reads like a stringified exception because that's more or less what it is, and it's worth routing through @nestjs/throttler's exception factory if this needs to match the rest of an API's error shape.

Where the throttle's job ends

None of that knows or cares who's logging in. Ten requests a minute from one IP against alice@example.com and ten requests a minute from ten different IPs, one attempt each, all against alice@example.com, look completely different to a per-IP counter — the second pattern never trips it, no matter how long it continues. That's the gap a second, IP-independent mechanism has to cover.

Locking the account, not the address

async recordFailure(email: string): Promise<FailureResult> {
  const key = this.attemptsKey(email);
  const attempts = await this.redis.incr(key);
  if (attempts === 1) {
    await this.redis.expire(key, this.windowSeconds);
  }

  if (attempts >= this.maxAttempts) {
    await this.redis.set(this.lockKey(email), '1', 'EX', this.lockoutSeconds);
    return { attempts, locked: true, retryAfterSeconds: this.lockoutSeconds };
  }

  return { attempts, locked: false };
}

async recordSuccess(email: string): Promise<void> {
  await this.redis.del(this.attemptsKey(email), this.lockKey(email));
}

INCR against a key that doesn't exist yet starts it at 1, so there's no separate setup step for a first failure. The counter's expiry is set once — only when attempts === 1 — so a streak of failures within one window doesn't keep pushing its own deadline out; it's one streak, with one expiry, not a self-renewing one.

Both key-building helpers lowercase the email first:

private attemptsKey(email: string): string {
  return `login-attempts:${email.toLowerCase()}`;
}

Skipping that normalization would open a real gap, not just a style nitpick — so rather than assume it works, this was tested directly: three failures as dana@example.com, then two more as Dana@Example.COM, a different casing entirely. The Redis counter read 5 afterward, under a single key, and a sixth attempt in yet another casing (DANA@EXAMPLE.COM) — this time with the correct password — still came back locked. Without the .toLowerCase(), those three casings would have been three separate five-strike budgets instead of one.

Two timers that look independent and aren't

The service has two separate durations: how long the failure counter itself lives (windowSeconds), and how long an actual lock lasts once triggered (lockoutSeconds). Treating them as unrelated knobs is the natural first instinct, and it's wrong — tested directly, with a 10-second window and a 6-second lock:

lock status right after the 6s lock expires: { locked: false }
attempts count at that same moment:          5

The lock is gone, but the counter — on its own, longer, 10-second clock — hasn't caught up yet. It's still sitting at 5. One more failed attempt right then doesn't start over at 1; it pushes the existing counter to 6, still over the 5-attempt threshold, and the account locks again immediately:

result of one failure right after unlock: { attempts: 6, locked: true }

So "unlocked" didn't mean "clean slate" here — it meant "one more mistake and you're back in." That might be exactly the behavior you want (harsher consequences for someone who fails again right after a lockout), but it should be a choice, not a side effect of two numbers that happened to get picked independently. Defaulting LOGIN_LOCKOUT_WINDOW_SECONDS and LOGIN_LOCKOUT_DURATION_SECONDS to the same value is what makes the out-of-the-box behavior a genuine reset: by the time the lock is gone, so is the counter it was based on. Tested with matched timers instead of mismatched ones, the positive case is exactly as boring as it should be — wait out the lock, and a correct password just works:

curl -X POST http://localhost:3000/auth/login -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"correct-horse-battery"}'
# -> 201 { "accessToken": "..." }   (no failures logged anywhere afterward)

The failure that triggers a lock doesn't announce it

try {
  const user = await this.authService.validatePassword(dto.email, dto.password);
  await this.loginAttemptService.recordSuccess(dto.email);
  return { accessToken: this.authService.issueToken(user) };
} catch (err) {
  await this.loginAttemptService.recordFailure(dto.email);
  throw err; // same error whether or not this failure just triggered a lock
}

Four wrong-password attempts return four identical 401s. The fifth — the one that actually crosses the threshold — returns that same 401, not a warning. The lock only becomes visible on whatever comes next, correct password included. Confirmed end to end: four 401s, a fifth 401 that locked the account behind the scenes, and only the sixth request got the 429. Nothing in that fifth response tells anyone it was the last try, which is one less signal an attacker gets to calibrate around.

Both mechanisms, same request

# alice fails 5 times, then even the right password is rejected
curl -X POST http://localhost:3000/auth/login -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"wrong-password"}'
# -> 401  (x5)

curl -X POST http://localhost:3000/auth/login -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"correct-horse-battery"}'
# -> 429 {"statusCode":429,"message":"Account temporarily locked...","reason":"ACCOUNT_LOCKED"}

# bob, meanwhile, is entirely unaffected
curl -X POST http://localhost:3000/auth/login -H "Content-Type: application/json" \
  -d '{"email":"bob@example.com","password":"correct-horse-battery"}'
# -> 201 { "accessToken": "..." }

The lockout 429 and the throttler's 429 share a status code but not a body — "reason":"ACCOUNT_LOCKED" versus the generic ThrottlerException message — so nothing downstream has to guess which mechanism fired.

One sequencing detail if you're replicating this yourself: the IP throttle counts every call to /login in its window, not just the ones in whatever you'd mentally group as "the throttle test." Running the lockout sequence above first, then immediately sending a batch of requests to check the throttle, tripped the 429 on the third request of that batch rather than somewhere near the tenth — because the eight requests just spent on the lockout scenario were already sitting in the same 60-second window. Not a bug, just a shared counter that doesn't know your test plan has phases.

Running it

Needs Postgres and Redis; Docker is the fastest path to both:

git clone <your-repo-url> && cd rate-limiting-nestjs-demo
npm install && cp .env.example .env
docker run --name rl-postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=ratelimit_demo -p 5432:5432 -d postgres:16
docker run --name rl-redis -p 6379:6379 -d redis:7
npm run start:dev

The shipped defaults (5 attempts, a 15-minute window and lock) are real production numbers, not demo placeholders, but they're still a starting point rather than a universal answer — how forgiving to be with a legitimate user who fumbles their password a few times is a judgment call specific to what's being protected. Lower LOGIN_LOCKOUT_WINDOW_SECONDS/LOGIN_LOCKOUT_DURATION_SECONDS in your own .env if you want to watch a lock expire without an actual 15-minute wait.

What's deliberately not here: the in-memory throttle storage this demo uses doesn't share counters across multiple app instances, so a horizontally-scaled deployment needs a shared store (@nestjs/throttler supports pluggable backends, Redis included) or the limit quietly stops meaning what it says. There's also no lockout notification to the account owner, and no CAPTCHA as a third layer for public-facing forms — both reasonable additions, both left out here to keep the two mechanisms that are built easy to see clearly.

The full project — this module, the lockout service, and a README with the complete setup and curl walkthrough — is in the rate-limiting-nestjs-demo repository alongside this post.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!