How Shadowbroker Handles Data Persistence and Schema Management in Its Backend

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, 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.

# 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. 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), 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.

# 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.

# 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, 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

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. 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. 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}