# How Shadowbroker Handles Data Persistence and Schema Management in Its Backend

> Discover how Shadowbroker manages data persistence and schema using its dual-layer architecture with in-memory storage and evolving SQLite databases for CCTV metadata.

- Repository: [Shadowbroker/Shadowbroker](https://github.com/BigBodyCobain/Shadowbroker)
- Tags: how-to-guide
- Published: 2026-05-07

---

**Shadowbroker implements a dual-layer storage architecture that combines a thread-safe in-memory dictionary for transient fetcher data with a SQLite database featuring programmatic schema evolution for durable CCTV metadata storage.**

Shadowbroker's backend employs a pragmatic approach to data persistence and schema management, balancing real-time performance requirements with operational simplicity. The system maintains two distinct storage mechanisms: an ephemeral in-memory store for sharing raw payloads between asynchronous fetcher jobs, and a persistent SQLite database for long-term camera metadata retention. According to the BigBodyCobain/Shadowbroker source code, this architecture avoids heavy migration frameworks while ensuring thread safety and ACID compliance across process restarts.

## Transient In-Memory Storage for Fetcher Modules

The system utilizes a **transient in-memory store** to enable communication between various "fetcher" modules without disk I/O overhead. Located in [`backend/tests/test_store.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/tests/test_store.py), this implementation consists of a simple Python dictionary named `latest_data` protected by a `threading.Lock` instance called `_data_lock`.

The lock guarantees thread-safe reads and writes when fetcher modules update shared state. The `latest_data` dictionary holds the most recent JSON structures keyed by source name (e.g., `cameras`, `weather`), allowing async jobs to access the latest raw payloads from any data source. The test suite verifies both the shape validation and thread-safety guarantees of this store.

```python

# Example: Access the transient in-memory store (used by fetchers)

from services.fetchers._store import _data_lock, latest_data

def some_fetcher():
    data = {"cameras": [{"id": "EX-001", "lat": 40.0, "lon": -74.0}]}
    with _data_lock:
        latest_data.update(data)   # thread-safe update

```

## Persistent SQLite Database for CCTV Pipeline

For durable storage, Shadowbroker utilizes a **single SQLite database** (`cctv.db` under the `data/` directory) managed through [`backend/services/cctv_pipeline.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/cctv_pipeline.py). This file handles all persistence logic for the CCTV ingest pipeline, including schema initialization, lightweight migrations, and data consistency operations.

### Database Initialization and Table Creation

When `init_db()` executes at service startup (called at line 96 of [`backend/services/cctv_pipeline.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/cctv_pipeline.py)), it creates the `cameras` table if it does not exist. The table schema includes columns for the camera identifier, source agency, geographic coordinates, direction facing, media URL, media type, refresh interval, and an auto-updated timestamp.

### Lightweight Schema Migration Strategy

Rather than employing a full migration framework like Alembic or Django migrations, Shadowbroker implements **programmatic schema evolution** directly within the initialization logic. The `init_db()` function queries the existing table structure using `PRAGMA table_info(cameras)` and checks for the presence of the `media_type` column. If this column is missing—indicating a database created before the column was added—the function executes an `ALTER TABLE` statement to append it (as implemented at line 158).

This pattern allows Shadowbroker to evolve its schema incrementally without external dependencies or complex versioning systems.

### UPSERT Operations for Data Consistency

Each concrete ingestor implementation (such as `TFLJamCamIngestor` and `GeorgiaDOTIngestor`) calls `BaseCCTVIngestor.ingest()`. The method establishes a database connection, optionally deletes stale entries for a specific source prefix, then performs an **UPSERT** operation using SQLite's `INSERT ... ON CONFLICT(id) DO UPDATE SET ...` syntax (lines 130-166).

This ensures atomic insertion of new camera records or refresh of existing ones while updating the `last_updated` timestamp to the current time.

```python

# Example: Upsert a camera record (run by each ingestor)

def ingest(self):
    conn = sqlite3.connect(str(DB_PATH))
    cursor = conn.cursor()
    for cam in self.fetch_data():
        cursor.execute(
            """
            INSERT INTO cameras
                (id, source_agency, lat, lon, direction_facing,
                 media_url, media_type, refresh_rate_seconds)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                source_agency=excluded.source_agency,
                lat=excluded.lat,
                lon=excluded.lon,
                direction_facing=excluded.direction_facing,
                media_url=excluded.media_url,
                media_type=excluded.media_type,
                refresh_rate_seconds=excluded.refresh_rate_seconds,
                last_updated=CURRENT_TIMESTAMP
            """,
            (
                cam["id"], cam["source_agency"], cam["lat"], cam["lon"],
                cam.get("direction_facing", "Unknown"),
                cam["media_url"], cam["media_type"], cam["refresh_rate_seconds"]
            ),
        )
    conn.commit()
    conn.close()

```

### Read-Only Access Patterns

The helper function `get_all_cameras()` (line 1225) opens a read-only connection to retrieve all camera rows. It normalizes the `media_type` column on-the-fly when necessary, providing safe concurrent access to metadata without blocking the UPSERT operations performed by ingestors.

```python

# Example: Initialise the CCTV database (called once at start-up)

from services.cctv_pipeline import init_db
init_db()                     # creates tables & adds missing columns

```

## Summary

- **Shadowbroker** combines an in-memory dictionary protected by `threading.Lock` for transient fetcher data with SQLite for persistent CCTV metadata storage
- Schema management occurs programmatically at startup via `init_db()` in [`backend/services/cctv_pipeline.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/cctv_pipeline.py), checking `PRAGMA table_info(cameras)` and running `ALTER TABLE` when columns like `media_type` are missing
- The **UPSERT** pattern (`INSERT ... ON CONFLICT(id) DO UPDATE`) in `BaseCCTVIngestor.ingest()` ensures idempotent data updates without duplicate entries
- All camera metadata survives process restarts through the `cctv.db` file, while fetcher modules share real-time data via the thread-safe `latest_data` store defined in [`backend/tests/test_store.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/tests/test_store.py)

## Frequently Asked Questions

### What database technology does Shadowbroker use for persistence?

Shadowbroker uses **SQLite** for all persistent storage, specifically a single database file named `cctv.db` stored in the `data/` directory. This lightweight approach eliminates external database dependencies while providing ACID guarantees for camera metadata through Python's built-in `sqlite3` module.

### How does Shadowbroker handle schema migrations without a dedicated framework?

The system implements **programmatic schema evolution** in the `init_db()` function within [`backend/services/cctv_pipeline.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/cctv_pipeline.py). On each service startup, the code inspects the existing table structure using `PRAGMA table_info(cameras)` and conditionally executes `ALTER TABLE` statements to add missing columns (such as `media_type`), avoiding complex migration tooling while supporting incremental upgrades.

### How do fetcher modules share data without writing to the database?

Fetcher modules utilize a **transient in-memory store** located in [`backend/tests/test_store.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/tests/test_store.py). This store consists of a global `latest_data` dictionary protected by `_data_lock` (a `threading.Lock` instance), allowing thread-safe sharing of JSON payloads between asynchronous jobs without disk I/O or database contention.

### What prevents duplicate camera records when multiple ingestors run simultaneously?

The `BaseCCTVIngestor.ingest()` method employs SQLite's **UPSERT** syntax (`INSERT ... ON CONFLICT(id) DO UPDATE SET ...`) to atomically insert new records or update existing ones based on the primary key. This ensures that the latest data from any ingestor (such as `TFLJamCamIngestor` or `GeorgiaDOTIngestor`) always wins while maintaining strict data consistency and updating the `last_updated` timestamp automatically.