DEV Community: Fekri

Build a small RAG app with Next.js, pgvector, and OpenAI

Fekri — Mon, 19 Oct 2026 20:06:40 +0000

Retrieval-augmented generation does not need a large framework or a separate vector database. This walkthrough builds a compact version with Next.js, Postgres, pgvector, the official OpenAI package, and pg.

The complete starter is here: https://github.com/buildfound/rag-starter

The flow has four parts:

Read Markdown and text files.
Split them into overlapping chunks and embed each chunk.
Retrieve the nearest chunks for a question with cosine distance.
Stream an answer that cites the numbered chunks.

Keep vectors in Postgres

If an application already needs relational data, pgvector keeps documents, metadata, and embeddings together. There is one database to run locally, one transaction model, and ordinary SQL for retrieval and filtering.

The local database runs from the pgvector Postgres 16 image. Its initialization SQL enables the extension and creates two tables:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
  id BIGSERIAL PRIMARY KEY,
  source TEXT NOT NULL UNIQUE,
  title TEXT NOT NULL
);

CREATE TABLE chunks (
  id BIGSERIAL PRIMARY KEY,
  document_id BIGINT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
  chunk_index INTEGER NOT NULL,
  content TEXT NOT NULL,
  embedding vector(1536) NOT NULL,
  UNIQUE (document_id, chunk_index)
);

CREATE INDEX chunks_embedding_hnsw_idx
  ON chunks USING hnsw (embedding vector_cosine_ops);

The vector size matches text-embedding-3-small with 1,536 dimensions. The HNSW index uses cosine distance, which is also used by the retrieval query.

Chunk near useful boundaries

Embedding an entire long document makes retrieval too coarse: a relevant paragraph can be buried inside unrelated text. Very small chunks lose context. This starter uses a deliberately simple middle ground: up to 1,200 characters, split near a paragraph boundary when possible, with roughly 200 characters repeated into the next chunk.

The core loop looks for a useful split late in the current window:

const target = Math.min(start + maxChars, text.length);
const minimumUsefulSplit = start + Math.floor(maxChars * 0.6);
const paragraphBreak = text.lastIndexOf("\n\n", target);
const wordBreak = text.lastIndexOf(" ", target);

if (paragraphBreak >= minimumUsefulSplit) {
  end = paragraphBreak;
} else if (wordBreak > start) {
  end = wordBreak;
}

After storing that slice, the next window begins around end - overlapChars. Whitespace normalization and progress checks keep the function predictable. It is a pure function, so it can be tested without a database or API key.

The ingestion command recursively finds .md and .txt files:

npm run ingest -- ./docs

It embeds chunks in batches of 100. A file path is the document's unique source. On re-ingestion, the script upserts the document, deletes its previous chunks, and inserts the new set in one transaction. That prevents old and new versions from appearing together in search results.

Retrieve with cosine distance

A question is embedded with the same model and dimensions as the documents. pgvector's <=> operator calculates cosine distance, so similarity is 1 - distance:

SELECT d.source, d.title, c.content,
       1 - (c.embedding <=> $1::vector) AS similarity
  FROM chunks c
  JOIN documents d ON d.id = c.document_id
 ORDER BY c.embedding <=> $1::vector
 LIMIT $2

The vector is a parameter, not interpolated SQL. The limit is also parameterized and bounded in application code.

This starter retrieves five chunks. That number is an application choice: adjust it after looking at the kind and length of your own documents.

Make citations part of the prompt

Each retrieved chunk is formatted with a stable number:

[1] How retrieval works (docs/how-it-works.md)
Documents are split into short, overlapping chunks...

The system instruction is explicit: answer only from the numbered context, treat that context as source material rather than instructions, cite factual claims with [n], and say when the answer is absent. This is small but important. Retrieval gives the model relevant text; the instruction defines how that text should be used.

Model-produced citations are still model output. For higher-assurance uses, add checks that citations exist, point to retrieved chunks, and support the claims beside them.

Stream text and send sources first

The route starts an OpenAI chat completion with stream: true, then converts each text delta to bytes for a standard ReadableStream:

const stream = new ReadableStream<Uint8Array>({
  async start(controller) {
    const encoder = new TextEncoder();
    for await (const event of completion) {
      const text = event.choices[0]?.delta?.content;
      if (text) controller.enqueue(encoder.encode(text));
    }
    controller.close();
  },
});

The answer body stays plain text. Source metadata needs to arrive before it, so the route places a base64-encoded JSON array in the x-sources response header. Each entry contains its number, file path, title, and similarity. The browser decodes the header immediately, reads the body stream, and displays only sources whose [n] markers appear in the answer.

An NDJSON stream could carry metadata in its first line, but the header keeps the answer body usable as plain text and makes the protocol easy to inspect.

Run it locally

Create the environment file, add an OpenAI API key, and run the stack:

docker compose up -d
cp .env.example .env.local
npm install
npm run ingest -- ./docs
npm run dev

The OpenAI clients and database pool are created lazily. As a result, npm run build can compile the application without a live database or API key; those values are required only when ingestion or a chat request runs.

From here, useful extensions include metadata filters, more input formats, authentication, rate limiting, and citation validation. The starter keeps those out of the base so the retrieval path remains easy to read.

Built by BuildFound, a founder-led AI product studio.

Keyword Rank Tracking in Python: iTunes Search API, SQLite and Cron, and the Walls You'll Hit (2026)

Fekri — Mon, 19 Oct 2026 09:23:37 +0000

If you ship an iOS app, sooner or later you want to know one boring thing every morning: where does my app show up when someone searches "calorie tracker"? Paid tools answer that, but the DIY version is small enough to build in an afternoon, and building it teaches you more about how App Store search data behaves than any dashboard will.

This post builds a daily keyword rank tracker in Python with:

Apple's public iTunes Search API (no key, no account),
SQLite for history (one file, nothing to run),
cron to run it every day,
a tiny report that shows what moved since the last run.

Then the useful part: the walls you hit once it runs for real.

All code below was run on Python 3.13 against the live API while writing this. It needs Python 3.10+ and requests.

The data source

The iTunes Search API is one public endpoint:

https://itunes.apple.com/search?term=calorie+tracker&country=us&entity=software&limit=200

entity=software restricts results to iOS apps, country picks the storefront, and limit maxes out at 200. Each result has a trackId (the numeric App Store id, the number in apps.apple.com/.../id341232718), the name, the seller, the rating count and more.

Our "rank" is simply the 1-based index of our app's trackId in that ordered list. Keep that definition in mind, because it comes back in the walls section.

Configuration

A JSON file holds what to track. Here I track two well-known calorie apps across two storefronts:

{
  "countries": ["us", "gb"],
  "apps": [6480417616, 341232718],
  "keywords": ["calorie tracker", "calorie counter", "food diary", "macro tracker"]
}

Track your own app plus two or three direct competitors. A rank on its own tells you very little. A rank next to your rivals' ranks tells you a lot.

The tracker

#!/usr/bin/env python3
"""Daily App Store keyword rank tracker: iTunes Search API + SQLite."""
import json
import random
import sqlite3
import sys
import time
from datetime import datetime, timezone

import requests

DB_PATH = "ranks.db"
SEARCH_URL = "https://itunes.apple.com/search"
MAX_RESULTS = 200          # the API's hard ceiling
MIN_INTERVAL = 3.5         # seconds between calls (~17/min, under Apple's ~20/min guidance)

SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
    observed_at TEXT NOT NULL,      -- UTC ISO timestamp
    country     TEXT NOT NULL,
    keyword     TEXT NOT NULL,
    app_id      INTEGER NOT NULL,
    position    INTEGER,            -- NULL = not in the top MAX_RESULTS
    result_count INTEGER NOT NULL,
    PRIMARY KEY (observed_at, country, keyword, app_id)
);
"""

_last_call = 0.0


def search(term: str, country: str, session: requests.Session, retries: int = 4) -> list[dict]:
    """Return the ordered result list for one search term, with polite pacing and backoff."""
    global _last_call
    params = {"term": term, "country": country, "entity": "software", "limit": MAX_RESULTS}
    for attempt in range(retries):
        wait = MIN_INTERVAL - (time.monotonic() - _last_call)
        if wait > 0:
            time.sleep(wait)
        _last_call = time.monotonic()
        resp = session.get(SEARCH_URL, params=params, timeout=20)
        if resp.status_code == 200:
            # Apple sometimes serves JSON as text/javascript; parse the body ourselves.
            return json.loads(resp.text).get("results", [])
        if resp.status_code in (403, 429, 500, 502, 503):
            backoff = (2 ** attempt) * 10 + random.uniform(0, 5)
            print(f"  {resp.status_code} for {term!r}, sleeping {backoff:.0f}s", file=sys.stderr)
            time.sleep(backoff)
            continue
        resp.raise_for_status()
    raise RuntimeError(f"giving up on {term!r} after {retries} attempts")


def position_of(app_id: int, results: list[dict]) -> int | None:
    for i, r in enumerate(results, start=1):
        if r.get("trackId") == app_id:
            return i
    return None


def run(config_path: str) -> None:
    cfg = json.load(open(config_path))
    db = sqlite3.connect(DB_PATH)
    db.executescript(SCHEMA)
    observed_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat()

    with requests.Session() as session:
        session.headers["User-Agent"] = "rank-tracker/0.1 (you@example.com)"
        for country in cfg["countries"]:
            for keyword in cfg["keywords"]:
                results = search(keyword, country, session)
                rows = [
                    (observed_at, country, keyword, app_id,
                     position_of(app_id, results), len(results))
                    for app_id in cfg["apps"]
                ]
                db.executemany("INSERT OR REPLACE INTO observations VALUES (?,?,?,?,?,?)", rows)
                db.commit()
                print(f"{country} {keyword!r}: " +
                      ", ".join(f"{a}=#{p}" if p else f"{a}=-" for _, _, _, a, p, _ in rows))
    db.close()


if __name__ == "__main__":
    run(sys.argv[1] if len(sys.argv) > 1 else "keywords.json")

A few choices worth explaining:

One timestamp per run. Every row from a run shares observed_at, so "compare today with yesterday" is a join on a single value, not a fuzzy time window.
NULL means "not found in the top 200", not zero. Storing 0 or 201 would poison every average you compute later.
result_count is stored too. If a search suddenly returns 12 results instead of 200, you want to know the API changed, not that your app dropped.
Commit after every keyword. If run 3 of 40 dies on a network error, the first 2 are still saved.

The report

#!/usr/bin/env python3
"""Compare the two most recent runs and print what moved."""
import sqlite3

QUERY = """
WITH runs AS (
    SELECT DISTINCT observed_at FROM observations ORDER BY observed_at DESC LIMIT 2
),
latest AS (SELECT MAX(observed_at) AS t FROM runs),
previous AS (SELECT MIN(observed_at) AS t FROM runs)
SELECT cur.country, cur.keyword, cur.app_id,
       prev.position AS before, cur.position AS after
FROM observations cur
JOIN latest ON cur.observed_at = latest.t
LEFT JOIN observations prev
       ON prev.observed_at = (SELECT t FROM previous)
      AND prev.country = cur.country
      AND prev.keyword = cur.keyword
      AND prev.app_id = cur.app_id
ORDER BY cur.country, cur.keyword, cur.app_id;
"""


def fmt(p):
    return f"#{p}" if p is not None else "-"


def main():
    db = sqlite3.connect("ranks.db")
    for country, keyword, app_id, before, after in db.execute(QUERY):
        if before == after:
            delta = "="
        elif before is None:
            delta = "NEW"
        elif after is None:
            delta = "LOST"
        else:
            delta = f"{before - after:+d}"   # positive = moved up
        print(f"{country:3} {keyword:28} {app_id:<12} {fmt(before):>5} -> {fmt(after):<5} {delta}")


if __name__ == "__main__":
    main()

Running the tracker twice and then the report gives output like this (trimmed):

us  calorie tracker   341232718     #1 -> #1    =
us  calorie tracker   6480417616    #3 -> #3    =
us  food diary        6480417616  #107 -> #107  =
gb  food diary        6480417616   #55 -> #55   =
gb  macro tracker     6480417616    #8 -> #8    =

Two runs a minute apart are boring, which is the point. Run it daily for two weeks and the +3 / LOST lines start telling a story.

Scheduling it

# every day at 06:17 server time
17 6 * * * cd /opt/rank-tracker && /usr/bin/python3 rank_tracker.py >> tracker.log 2>&1 && /usr/bin/python3 report.py >> report.log

Pick an odd minute. Everyone's cron fires at :00, and you don't need to be part of that crowd. Since timestamps are stored in UTC, moving servers or time zones won't scramble your history.

That's the whole tracker. Now the walls.

Wall 1: The Search API is not the App Store search box

This matters most. The iTunes Search API searches the store catalogue. Nobody promises its ordering matches what a person sees in the App Store app. The real search results page can include a Search Ads slot, editorial modules and possibly per-user differences, and Apple doesn't document how either ranking works.

So treat your numbers as a consistent proxy, not ground truth. Measured the same way every day, the trend is meaningful even if "#3" isn't exactly what your users see. Spot-check a handful of keywords on a real device now and then, and write down when the two disagree.

Wall 2: Rate limits are real and vague

Apple's documentation describes the limit as roughly 20 calls per minute, subject to change. Exceed it and you get 403s. That's why the script spaces calls 3.5 s apart and backs off exponentially with jitter on 403/429/5xx.

The arithmetic matters: 50 keywords Ã— 5 storefronts = 250 calls, which is about 15 minutes at a polite pace. Fine for cron, painful if you planned to call this from a web request. If you need more throughput, more IPs is the wrong answer. Track fewer, better keywords.

Wall 3: The 200-result ceiling

limit stops at 200. If your app sits at #240 for a broad term, the API can't tell you that. You only learn "not in the top 200". For a new app that's most keywords, so start by tracking long-tail phrases where you actually appear. A keyword list full of NULLs is a sign to pick different keywords, not to write a smarter scraper.

Wall 4: Every storefront is a different world

The same phrase ranks differently per country. In the run above, one app sat at #107 for "food diary" in the US and #55 in the UK on the same morning. The country parameter changes the results, and so does the language people search in. Don't average storefronts together. Report them separately, and only track storefronts where you actually have users.

Wall 5: Response quirks

Small things that bite:

Some responses come back as text/javascript, not application/json. Parse the body yourself and don't let a strict client reject it.
Results for the same query can wobble between calls. One observation per keyword per day, stored with its timestamp, is more honest than polling hourly and chasing noise.
Apps get removed, renamed or re-categorised. Always key on trackId, never on the app name.

Wall 6: Terms of service and the scraping temptation

Once the Search API's limits annoy you, scraping apps.apple.com or the private storefront endpoints the App Store app uses will look tempting. Don't build on that. Those endpoints are undocumented, change without notice, sit behind bot protection, and using them may break Apple's terms. Read Apple's terms for the Search API too before you build anything commercial on top of it. A tracker that quietly breaks in month three is worse than one that tracks fewer keywords honestly.

Where this stops scaling

The script answers "where do I rank for the words I already chose?" It can't answer the more useful question: which words should I be tracking at all? You can only check keywords you thought to type, so the competitor ranking well for a phrase you never considered stays invisible.

That keyword discovery step is where I stopped writing scripts. I use PeekPanda*, which shows the App Store searches an iOS app has been observed in, with the date each position was seen, plus a keyword gap view comparing your app with a rival. I still keep this little tracker running, because a cron job and a SQLite file you fully understand make a good sanity check against any tool.

Wrap-up

About 120 lines of Python gives you a daily, versioned history of App Store keyword positions.
Store NULL for "not found", one timestamp per run, and key everything on trackId.
Treat positions as a proxy, respect the rate limit, keep storefronts separate, and leave private endpoints alone.

If you run this for a month, the most interesting thing you'll find probably won't be your own rank. It'll be watching a competitor slowly climb a keyword you'd ignored.

*I build it.