Test RAG Webhooks Without Exposing Your Vector DB

Use an ingestion boundary, not a public vector database

A webhook sender requires an HTTP endpoint that can accept events. It does not need access to a vector database listener, embedding service, queue administration page, container dashboard, or retrieval API. Publishing those internal services would expand the attack surface and couple an untrusted external payload directly to storage.

The public component in this tutorial is a small FastAPI receiver. It validates GitHub's documented signature against the exact request body, checks required GitHub headers, accepts only selected event types, validates a minimal schema, and stores the event in SQLite before returning success. A background worker reads durable pending jobs and updates Qdrant through its loopback-only API.

GitHub Issues webhook
        |
        | HTTPS POST
        v
Localtonet public URL
        |
        v
FastAPI receiver on 127.0.0.1:8000
  | verify X-Hub-Signature-256
  | validate X-GitHub-Delivery
  | allow selected events and actions
  | insert event into SQLite transaction
  v
SQLite durable queue
        |
        v
Background ingestion worker
  | normalize issue content
  | create deterministic chunks
  | generate local test embeddings
  | replace or delete document vectors
  v
Qdrant on 127.0.0.1:6333

The example uses GitHub Issues because it supplies a named provider contract, a unique delivery identifier, a documented SHA-256 webhook signature, and realistic create, edit, and delete mutations. Each issue becomes one logical RAG document. The repository name forms the authorization scope used during retrieval.

The embedding function is deliberately local and deterministic. It hashes normalized tokens into a fixed-size vector so the tutorial can run without an external model API or model credential. It is suitable for testing delivery, durability, replacement, deletion, metadata filtering, and retrieval plumbing. It is not presented as a production-quality semantic embedding model. Replace it with your approved embedding model only after the boundary and failure behavior work correctly.

Prerequisites and selected example stack

This tutorial uses one concrete stack so you can complete it without choosing your own framework, queue, embedding implementation, or vector store. The commands assume a shell that can create a Python virtual environment and run Docker Compose. On Windows, use PowerShell equivalents where activation syntax differs.

Create an empty working directory and move into it:

mkdir local-rag-webhook
cd local-rag-webhook

The completed project has this layout:

local-rag-webhook/
├── .env
├── .env.example
├── .gitignore
├── docker-compose.yml
├── main.py
├── query.py
├── requirements.txt
├── send_test.py
└── data/
    ├── events.db
    └── qdrant/

Use GitHub's exact webhook contract

Signature handling and retry behavior are provider-specific. This implementation follows GitHub's webhook contract rather than presenting a generic HMAC format as universal.

GitHub sends the SHA-256 signature in X-Hub-Signature-256. The value begins with sha256= and contains a hexadecimal HMAC calculated from the exact request body and the configured webhook secret. GitHub's primary documentation instructs receivers to use a secure constant-time comparison and warns that the payload and secret must be handled as UTF-8. See GitHub's webhook signature validation documentation for the provider's current rules and examples.

GitHub also supplies X-GitHub-Delivery, which identifies the delivery, and X-GitHub-Event, which names the event. This example uses the delivery ID as the receiver-level idempotency key. It accepts the ping event for webhook setup checks and selected actions from the issues event.

GitHub recommends returning a 2xx response within ten seconds. It does not automatically redeliver failed webhook deliveries, so application retries and provider retries must not be conflated. Failed deliveries can be redelivered through GitHub's supported redelivery controls. Review GitHub's webhook best practices and GitHub's webhook redelivery documentation before relying on current delivery-management behavior.

Why there is no arbitrary freshness window

Some providers sign a timestamp together with the payload. GitHub's documented X-Hub-Signature-256 construction signs the payload body with the webhook secret. This tutorial therefore does not pretend that an unsigned request time is a trustworthy replay defense.

A captured valid delivery could theoretically be replayed, but its X-GitHub-Delivery value is accepted only once by the SQLite uniqueness constraint. A request that changes the delivery ID without recalculating the signature can still have a valid body signature because the delivery header is not part of GitHub's documented HMAC input. For a high-risk deployment, supplement provider authentication with network restrictions or another trusted gateway control where supported. Do not modify GitHub's signature formula.

Accepted event mapping

Build the receiver, durable queue, and indexer

1. Install the Python dependencies

Create requirements.txt:

fastapi
uvicorn[standard]
qdrant-client
python-dotenv
httpx

Create and activate a virtual environment, then install the packages:

python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -r requirements.txt

In PowerShell, activate the environment with:

.venv\Scripts\Activate.ps1

2. Configure loopback-only Qdrant

Create docker-compose.yml:

services:
  qdrant:
    image: qdrant/qdrant:latest
    restart: unless-stopped
    ports:
      - "127.0.0.1:6333:6333"
    volumes:
      - ./data/qdrant:/qdrant/storage

The host-side address is explicitly 127.0.0.1:6333. It is not published on every network interface. For a maintained team environment, review and pin an approved Qdrant image version instead of allowing an unreviewed image update.

3. Create safe environment files

Create .env.example:

GITHUB_WEBHOOK_SECRET=replace-with-a-random-test-secret
SQLITE_PATH=./data/events.db
QDRANT_URL=http://127.0.0.1:6333
QDRANT_COLLECTION=github_issue_chunks
MAX_BODY_BYTES=1048576

Copy it to .env and generate a secret locally:

cp .env.example .env
python -c "import secrets; print(secrets.token_hex(32))"

Replace the placeholder in .env with the generated value. Use that same secret when configuring the GitHub webhook. Do not paste the secret into this article, a repository, screenshots, command history shared with others, or application logs.

Create .gitignore:

.venv/
.env
__pycache__/
data/

4. Create the FastAPI receiver and durable worker

Create main.py. The request path reads raw bytes before JSON parsing, performs constant-time signature comparison, normalizes only the fields required by the worker, and inserts the delivery into SQLite. The worker recovers interrupted processing jobs at startup, retries transient failures, and uses deterministic vector IDs.

import hashlib
import hmac
import json
import math
import os
import re
import sqlite3
import threading
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    FieldCondition,
    Filter,
    MatchValue,
    PointStruct,
    VectorParams,
)

load_dotenv()

SECRET = os.getenv("GITHUB_WEBHOOK_SECRET", "")
SQLITE_PATH = os.getenv("SQLITE_PATH", "./data/events.db")
QDRANT_URL = os.getenv("QDRANT_URL", "http://127.0.0.1:6333")
COLLECTION = os.getenv("QDRANT_COLLECTION", "github_issue_chunks")
MAX_BODY_BYTES = int(os.getenv("MAX_BODY_BYTES", "1048576"))
VECTOR_SIZE = 384
MAX_ATTEMPTS = 5
STOP = threading.Event()

if not SECRET or SECRET == "replace-with-a-random-test-secret":
    raise RuntimeError("Set a real GITHUB_WEBHOOK_SECRET in .env")

Path(SQLITE_PATH).parent.mkdir(parents=True, exist_ok=True)
qdrant = QDRANTClient = QdrantClient(url=QDRANT_URL)

def db():
    connection = sqlite3.connect(SQLITE_PATH, timeout=30)
    connection.row_factory = sqlite3.Row
    return connection

def initialize_storage():
    with db() as connection:
        connection.execute("PRAGMA journal_mode=WAL")
        connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS deliveries (
                delivery_id TEXT PRIMARY KEY,
                event_name TEXT NOT NULL,
                action TEXT NOT NULL,
                payload_json TEXT NOT NULL,
                status TEXT NOT NULL,
                attempts INTEGER NOT NULL DEFAULT 0,
                next_attempt_at REAL NOT NULL DEFAULT 0,
                last_error TEXT,
                created_at REAL NOT NULL,
                completed_at REAL
            );

            CREATE TABLE IF NOT EXISTS document_state (
                issue_key TEXT PRIMARY KEY,
                source_version TEXT NOT NULL,
                deleted INTEGER NOT NULL DEFAULT 0
            );
            """
        )
        connection.execute(
            """
            UPDATE deliveries
            SET status = 'pending',
                next_attempt_at = 0,
                last_error = 'Recovered after worker restart'
            WHERE status = 'processing'
            """
        )

    if not qdrant.collection_exists(COLLECTION):
        qdrant.create_collection(
            collection_name=COLLECTION,
            vectors_config=VectorParams(
                size=VECTOR_SIZE,
                distance=Distance.COSINE,
            ),
        )

def verify_signature(raw_body: bytes, supplied: str) -> bool:
    if not supplied.startswith("sha256="):
        return False
    expected = "sha256=" + hmac.new(
        SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, supplied)

def normalize_issue_event(payload: dict) -> dict:
    action = payload.get("action")
    repository = payload.get("repository") or {}
    issue = payload.get("issue") or {}

    if action not in {"opened", "edited", "deleted"}:
        raise ValueError("Unsupported issues action")

    required = {
        "repository.full_name": repository.get("full_name"),
        "issue.id": issue.get("id"),
        "issue.number": issue.get("number"),
        "issue.updated_at": issue.get("updated_at"),
    }
    missing = [name for name, value in required.items() if value is None]
    if missing:
        raise ValueError("Missing fields: " + ", ".join(missing))

    return {
        "action": action,
        "repo": str(repository["full_name"]),
        "issue_id": str(issue["id"]),
        "issue_number": int(issue["number"]),
        "source_version": str(issue["updated_at"]),
        "title": str(issue.get("title") or ""),
        "body": str(issue.get("body") or ""),
        "html_url": str(issue.get("html_url") or ""),
    }

def tokenize_embedding(text: str) -> list[float]:
    vector = [0.0] * VECTOR_SIZE
    tokens = re.findall(r"[a-z0-9_]+", text.lower())

    for token in tokens:
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        index = int.from_bytes(digest[:4], "big") % VECTOR_SIZE
        sign = 1.0 if digest[4] % 2 == 0 else -1.0
        vector[index] += sign

    norm = math.sqrt(sum(value * value for value in vector))
    if norm == 0:
        return vector
    return [value / norm for value in vector]

def split_text(text: str, size: int = 120, overlap: int = 20) -> list[str]:
    words = text.split()
    if not words:
        return []

    chunks = []
    start = 0
    while start < len(words):
        end = min(start + size, len(words))
        chunks.append(" ".join(words[start:end]))
        if end == len(words):
            break
        start = max(end - overlap, start + 1)
    return chunks

def issue_filter(issue_key: str) -> Filter:
    return Filter(
        must=[
            FieldCondition(
                key="issue_key",
                match=MatchValue(value=issue_key),
            )
        ]
    )

def current_version(issue_key: str):
    with db() as connection:
        return connection.execute(
            """
            SELECT source_version, deleted
            FROM document_state
            WHERE issue_key = ?
            """,
            (issue_key,),
        ).fetchone()

def save_document_state(issue_key: str, version: str, deleted: bool):
    with db() as connection:
        connection.execute(
            """
            INSERT INTO document_state(issue_key, source_version, deleted)
            VALUES (?, ?, ?)
            ON CONFLICT(issue_key) DO UPDATE SET
                source_version = excluded.source_version,
                deleted = excluded.deleted
            """,
            (issue_key, version, 1 if deleted else 0),
        )

def process_event(event: dict):
    issue_key = f"{event['repo']}:{event['issue_id']}"
    existing = current_version(issue_key)

    if existing and event["source_version"] < existing["source_version"]:
        return

    if event["action"] == "deleted":
        qdrant.delete(
            collection_name=COLLECTION,
            points_selector=issue_filter(issue_key),
            wait=True,
        )
        save_document_state(issue_key, event["source_version"], True)
        return

    document = (event["title"] + "\n\n" + event["body"]).strip()
    chunks = split_text(document)

    qdrant.delete(
        collection_name=COLLECTION,
        points_selector=issue_filter(issue_key),
        wait=True,
    )

    points = []
    for position, chunk in enumerate(chunks):
        identity = (
            f"{issue_key}:{event['source_version']}:{position}"
        )
        point_id = str(uuid.uuid5(uuid.NAMESPACE_URL, identity))
        points.append(
            PointStruct(
                id=point_id,
                vector=tokenize_embedding(chunk),
                payload={
                    "issue_key": issue_key,
                    "repo": event["repo"],
                    "issue_id": event["issue_id"],
                    "issue_number": event["issue_number"],
                    "source_version": event["source_version"],
                    "chunk_position": position,
                    "text": chunk,
                    "html_url": event["html_url"],
                },
            )
        )

    if points:
        qdrant.upsert(
            collection_name=COLLECTION,
            points=points,
            wait=True,
        )

    save_document_state(issue_key, event["source_version"], False)

def claim_job():
    now = time.time()
    connection = db()
    try:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT *
            FROM deliveries
            WHERE status IN ('pending', 'retrying')
              AND attempts < ?
              AND next_attempt_at <= ?
            ORDER BY created_at
            LIMIT 1
            """,
            (MAX_ATTEMPTS, now),
        ).fetchone()

        if row is None:
            connection.commit()
            return None

        connection.execute(
            """
            UPDATE deliveries
            SET status = 'processing',
                attempts = attempts + 1
            WHERE delivery_id = ?
            """,
            (row["delivery_id"],),
        )
        connection.commit()
        return dict(row)
    finally:
        connection.close()

def mark_complete(delivery_id: str):
    with db() as connection:
        connection.execute(
            """
            UPDATE deliveries
            SET status = 'completed',
                completed_at = ?,
                last_error = NULL
            WHERE delivery_id = ?
            """,
            (time.time(), delivery_id),
        )

def mark_failed(delivery_id: str, attempts: int, error: Exception):
    next_attempt = time.time() + min(2 ** attempts, 60)
    status = "failed" if attempts + 1 >= MAX_ATTEMPTS else "retrying"

    with db() as connection:
        connection.execute(
            """
            UPDATE deliveries
            SET status = ?,
                next_attempt_at = ?,
                last_error = ?
            WHERE delivery_id = ?
            """,
            (
                status,
                next_attempt,
                str(error)[:500],
                delivery_id,
            ),
        )

def worker_loop():
    while not STOP.is_set():
        job = claim_job()
        if job is None:
            STOP.wait(0.5)
            continue

        try:
            process_event(json.loads(job["payload_json"]))
            mark_complete(job["delivery_id"])
        except Exception as error:
            mark_failed(job["delivery_id"], job["attempts"], error)

@asynccontextmanager
async def lifespan(app: FastAPI):
    initialize_storage()
    STOP.clear()
    worker = threading.Thread(target=worker_loop, daemon=True)
    worker.start()
    yield
    STOP.set()
    worker.join(timeout=5)

app = FastAPI(
    title="Local RAG GitHub Webhook Receiver",
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
    lifespan=lifespan,
)

@app.get("/healthz")
def health():
    return {"status": "ready"}

@app.get("/queue-status")
def queue_status():
    with db() as connection:
        rows = connection.execute(
            """
            SELECT status, COUNT(*) AS count
            FROM deliveries
            GROUP BY status
            ORDER BY status
            """
        ).fetchall()
    return {row["status"]: row["count"] for row in rows}

@app.post("/webhooks/github")
async def github_webhook(request: Request):
    content_type = request.headers.get("content-type", "")
    if "application/json" not in content_type.lower():
        raise HTTPException(415, "Expected application/json")

    declared_length = request.headers.get("content-length")
    if declared_length and int(declared_length) > MAX_BODY_BYTES:
        raise HTTPException(413, "Request body too large")

    raw_body = await request.body()
    if len(raw_body) > MAX_BODY_BYTES:
        raise HTTPException(413, "Request body too large")

    signature = request.headers.get("x-hub-signature-256", "")
    if not verify_signature(raw_body, signature):
        raise HTTPException(401, "Invalid webhook signature")

    event_name = request.headers.get("x-github-event", "")
    delivery_id = request.headers.get("x-github-delivery", "")
    if not event_name or not delivery_id:
        raise HTTPException(400, "Missing GitHub delivery headers")

    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError:
        raise HTTPException(400, "Invalid JSON")

    if event_name == "ping":
        return {"status": "pong"}

    if event_name != "issues":
        return {"status": "ignored", "event": event_name}

    try:
        normalized = normalize_issue_event(payload)
    except ValueError as error:
        if "Unsupported issues action" in str(error):
            return {
                "status": "ignored",
                "action": payload.get("action"),
            }
        raise HTTPException(422, str(error))

    try:
        with db() as connection:
            connection.execute(
                """
                INSERT INTO deliveries(
                    delivery_id,
                    event_name,
                    action,
                    payload_json,
                    status,
                    attempts,
                    next_attempt_at,
                    created_at
                )
                VALUES (?, ?, ?, ?, 'pending', 0, 0, ?)
                """,
                (
                    delivery_id,
                    event_name,
                    normalized["action"],
                    json.dumps(normalized, separators=(",", ":")),
                    time.time(),
                ),
            )
    except sqlite3.IntegrityError:
        return {
            "status": "duplicate",
            "delivery_id": delivery_id,
        }

    return {
        "status": "accepted",
        "delivery_id": delivery_id,
    }

The unusual assignment to QDRANTClient is unnecessary, so simplify that line before running:

qdrant = QdrantClient(url=QDRANT_URL)

This correction is called out explicitly so the final file contains only the simple assignment above. Do not leave the intermediate alias in the file.

5. Create the local signed-delivery tool

Create send_test.py. It produces GitHub-compatible test headers, signs the exact serialized body, and lets you reuse a delivery ID to verify deduplication.

import argparse
import hashlib
import hmac
import json
import os
import uuid
from datetime import datetime, timezone

import httpx
from dotenv import load_dotenv

load_dotenv()
SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--url",
        default="http://127.0.0.1:8000/webhooks/github",
    )
    parser.add_argument(
        "--action",
        choices=["opened", "edited", "deleted"],
        default="opened",
    )
    parser.add_argument("--delivery-id", default=str(uuid.uuid4()))
    parser.add_argument("--invalid-signature", action="store_true")
    parser.add_argument("--issue-id", default="9001")
    parser.add_argument("--issue-number", type=int, default=1)
    parser.add_argument(
        "--title",
        default="Reset a local development API token",
    )
    parser.add_argument(
        "--body",
        default=(
            "Rotate the test token in the developer settings page. "
            "Never commit the replacement token to source control."
        ),
    )
    arguments = parser.parse_args()

    payload = {
        "action": arguments.action,
        "repository": {
            "full_name": "example/private-rag-test"
        },
        "issue": {
            "id": int(arguments.issue_id),
            "number": arguments.issue_number,
            "title": arguments.title,
            "body": arguments.body,
            "html_url": (
                "https://github.com/example/private-rag-test/issues/1"
            ),
            "updated_at": datetime.now(timezone.utc).isoformat(),
        },
    }

    raw_body = json.dumps(
        payload,
        separators=(",", ":"),
    ).encode("utf-8")

    signature = "sha256=" + hmac.new(
        SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    if arguments.invalid_signature:
        signature = "sha256=" + ("0" * 64)

    response = httpx.post(
        arguments.url,
        content=raw_body,
        headers={
            "Content-Type": "application/json",
            "X-GitHub-Event": "issues",
            "X-GitHub-Delivery": arguments.delivery_id,
            "X-Hub-Signature-256": signature,
        },
        timeout=10,
    )

    print("status:", response.status_code)
    print("delivery_id:", arguments.delivery_id)
    print("body:", response.text)

if __name__ == "__main__":
    main()

6. Create the repository-filtered retrieval client

Retrieval stays local. Create query.py:

import argparse
import hashlib
import math
import os
import re

from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.models import FieldCondition, Filter, MatchValue

load_dotenv()

QDRANT_URL = os.getenv(
    "QDRANT_URL",
    "http://127.0.0.1:6333",
)
COLLECTION = os.getenv(
    "QDRANT_COLLECTION",
    "github_issue_chunks",
)
VECTOR_SIZE = 384

def embedding(text: str) -> list[float]:
    vector = [0.0] * VECTOR_SIZE
    for token in re.findall(r"[a-z0-9_]+", text.lower()):
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        index = int.from_bytes(digest[:4], "big") % VECTOR_SIZE
        sign = 1.0 if digest[4] % 2 == 0 else -1.0
        vector[index] += sign

    norm = math.sqrt(sum(value * value for value in vector))
    return vector if norm == 0 else [value / norm for value in vector]

parser = argparse.ArgumentParser()
parser.add_argument("--repo", required=True)
parser.add_argument("--query", required=True)
arguments = parser.parse_args()

client = QdrantClient(url=QDRANT_URL)
result = client.query_points(
    collection_name=COLLECTION,
    query=embedding(arguments.query),
    query_filter=Filter(
        must=[
            FieldCondition(
                key="repo",
                match=MatchValue(value=arguments.repo),
            )
        ]
    ),
    limit=5,
    with_payload=True,
)

if not result.points:
    print("No authorized matches.")
else:
    for point in result.points:
        print(f"score={point.score:.4f}")
        print(f"issue={point.payload['issue_key']}")
        print(point.payload["text"])
        print()

Start the complete local stack

docker compose up -d
curl http://127.0.0.1:6333/collections

source .venv/bin/activate
uvicorn main:app --host 127.0.0.1 --port 8000

Keep the Uvicorn process running in its terminal. Use a second terminal, activate the same virtual environment, and run the verification commands below.

Verify signatures, durability, indexing, retrieval, and deletion locally

Do not create a public tunnel until all local tests pass. Each check below has a concrete expected outcome.

1. Accept one correctly signed delivery

python send_test.py --delivery-id local-delivery-001

Expected result:

status: 200
delivery_id: local-delivery-001
body: {"status":"accepted","delivery_id":"local-delivery-001"}

Wait briefly for the worker, then inspect queue state:

curl http://127.0.0.1:8000/queue-status

The response should contain one completed delivery:

{"completed":1}

2. Reject an invalid signature

python send_test.py \
  --delivery-id invalid-delivery-001 \
  --invalid-signature

Expected result:

status: 401
body: {"detail":"Invalid webhook signature"}

Run the queue-status request again. The count must not increase. Authentication happens before JSON processing or durable insertion.

3. Treat a repeated delivery as a duplicate

Send the same delivery ID again:

python send_test.py --delivery-id local-delivery-001

Expected result:

status: 200
delivery_id: local-delivery-001
body: {"status":"duplicate","delivery_id":"local-delivery-001"}

The completed count remains unchanged. Returning a successful duplicate response prevents a legitimate manual redelivery from being mistaken for a receiver failure while ensuring that it does not create a second job.

4. Confirm authorized retrieval

python query.py \
  --repo example/private-rag-test \
  --query "How should the development token be rotated?"

Expected output includes the indexed issue key and text referring to rotating a test token. The numeric score can vary with the query text, so the useful assertion is that the expected issue appears under the permitted repository filter.

Now use a repository scope that was not indexed:

python query.py \
  --repo another-team/repository \
  --query "development token"

Expected result:

No authorized matches.

This is a functional authorization-isolation test. In a real application, the repository or tenant scope must come from the authenticated caller's permissions, not from an unrestricted query parameter.

5. Verify an update replaces the active document

python send_test.py \
  --action edited \
  --delivery-id local-delivery-002 \
  --body "Rotate the sandbox credential and store it only in the approved secret manager."

Wait for completion, then query for approved secret manager. The new text should be returned. Querying the distinctive old phrase should no longer return an old active chunk for the same issue.

6. Verify deletion removes every issue chunk

python send_test.py \
  --action deleted \
  --delivery-id local-delivery-003

After the job completes, run:

python query.py \
  --repo example/private-rag-test \
  --query "approved secret manager"

Expected result:

No authorized matches.

Deletion is document-scoped. The worker removes every point carrying the logical issue key, then stores a tombstone version in SQLite so an older queued update cannot automatically become active afterward.

7. Verify queue persistence across a worker restart

Stop Qdrant while leaving the receiver running:

docker compose stop qdrant

Submit a valid event:

python send_test.py --delivery-id persistence-test-001

The receiver should still return accepted because SQLite durably stores the event. The worker cannot reach Qdrant, so queue status should show the job as retrying or eventually failed, depending on how long Qdrant remains stopped.

Restart Qdrant before the maximum attempt count is exhausted:

docker compose start qdrant

The worker should retry and move the event to completed. If you also stop Uvicorn while a job is marked processing, startup recovery returns that job to pending. Because the vector identities are deterministic, processing it again converges on the same point IDs rather than adding a duplicate set.

Expose only the receiver with a Localtonet HTTP tunnel

After local verification succeeds, Localtonet can provide the public HTTPS address GitHub requires. Our client runs on the machine that can reach FastAPI and establishes an outbound connection to a Localtonet relay server. You do not need inbound router port forwarding, inbound firewall changes, VPN setup, or a public IP address.

Follow the current installation path in our HTTP tunnel documentation. The available operating-system package, client instructions, relay choices, and dashboard options can change, so obtain those current values from the official page and dashboard rather than copying an old command.

Configure the HTTP tunnel in the documented sequence

If Localtonet cannot reach the receiver, verify that the client runs in an environment that can connect to the configured target. Containers, virtual machines, and subsystem environments can have separate loopback interfaces. Test reachability from the client environment itself.

Configure the GitHub repository webhook

In the test repository's webhook settings, add a repository webhook with these values:

• Payload URL: the Localtonet HTTPS address followed by /webhooks/github
• Content type: application/json
• Secret: the exact value stored in your local .env
• Events: select individual events and enable Issues
• Active: enabled only for the intended test period

GitHub normally sends a ping event when the webhook is created. A successful setup request should return:

{"status":"pong"}

Verify a real public delivery

Create a synthetic issue in the configured repository. GitHub should send an issues event through the Localtonet URL. Confirm all of the following:

• The GitHub delivery view reports a successful 2xx response.
• The receiver returns an accepted delivery ID.
• /queue-status moves the delivery to completed.
• A local query.py request using the exact repository scope returns the expected issue text.
• The same query under an unrelated repository scope returns no authorized matches.

Edit the issue and verify replacement. If your repository permissions and workflow allow issue deletion, delete the synthetic issue and verify that its distinctive content is no longer retrievable. Otherwise, retain the reproducible local deletion test as the deletion-path validation and remove the test issue according to your repository policy.

Test provider redelivery without creating duplicate work

Use GitHub's supported redelivery control for the successful Issues delivery. GitHub should resend the same delivery identity. The receiver should return a duplicate response, and the SQLite completed count should not increase for that delivery.

Do not intentionally make the public endpoint return errors merely to wait for an automatic retry. GitHub does not automatically redeliver failed webhook deliveries. Use the provider's documented manual redelivery workflow when testing this behavior.

Understand the remaining security boundary

HTTPS protects transport to the public endpoint, but it does not replace webhook authentication. The application still has to decide whether the body is authentic, whether the event type is allowed, whether the source is authorized to mutate the selected collection, and whether the payload is safe to retain.

Metadata is part of authorization

A relevant vector result can still be unauthorized. Repository, tenant, workspace, document classification, and user-permission metadata must be applied as mandatory filters during retrieval. Do not retrieve across all vectors and ask a language model to decide which chunks the caller was allowed to see.

The command-line query in this tutorial requires a repository value, but it does not authenticate a user. It demonstrates filtered vector retrieval, not a complete authorization service. A production retrieval API must derive permitted scopes from a validated identity and reject caller-supplied scopes that exceed those permissions.

Raw-body handling must remain exact

Signature verification occurs before parsing because parsed JSON cannot be safely serialized back into the original byte sequence. Whitespace, key ordering, escaping, and Unicode handling may change. The receiver signs the original bytes from await request.body() and only parses the body after authentication succeeds.

Queue durability does not eliminate poison jobs

A malformed but authentic event, dependency incompatibility, or unexpected vector-store failure can still prevent completion. The worker uses bounded retries and records a sanitized error. After the maximum attempt count, the job becomes failed instead of retrying forever.

For a larger system, move failed jobs into a dedicated quarantine workflow and require an authorized operator to replay them. Keep any replay function private. Replaying old raw HTTP requests through the public endpoint is not always appropriate because provider credentials, payload sensitivity, and event state may have changed.

Operate and troubleshoot the ingestion path

Monitor ingress separately from eventual indexing

A 2xx response means an authentic event was durably accepted or recognized as a duplicate. It does not mean embedding and indexing finished. Monitor at least the counts of pending, processing, retrying, completed, and failed deliveries. Queue age is especially important because a healthy receiver can continue accepting events while an unavailable worker falls behind.

Useful operational records include the delivery ID, event name, action, repository, issue identifier, validation outcome, queue status, attempt count, completion time, and sanitized error. Avoid retaining full issue bodies in ordinary logs.

Troubleshoot from the public boundary inward

Know the limitations of this compact example

SQLite is appropriate for a reproducible local test but is not automatically the right queue for a distributed production service. The worker and receiver share one machine. The Qdrant mutation and SQLite state update are not one cross-system transaction. Recovery relies on idempotent, repeatable operations.

GitHub's issue updated_at value is used as a source version. If two meaningful mutations have the same provider timestamp granularity, a more complete implementation may need an additional ordering rule. Repository renames also require a stable repository identifier rather than relying only on full_name.

The hashing vectorizer is designed to make the tutorial self-contained. Evaluate an approved semantic embedding model using a fixed retrieval test set before making claims about RAG quality. Record the model identifier, vector dimension, chunker version, normalization rules, and source revision in metadata when you introduce a real embedding model.

Shut down public access and remove test data

Tunnel setup and cleanup are separate lifecycle phases. Do not append shutdown as an extra configuration step or assume that deleting a provider webhook automatically stops the tunnel.

docker compose down
rm -rf data
rm -f .env

Use platform-appropriate deletion commands on Windows. Before removing data, confirm that no approved troubleshooting, audit, or retention requirement applies.

Frequently asked questions