# Sightline - ALPR / Flock Intelligence Platform

**Design & Architecture Document**

Author: (portfolio project) · Status: MVP build · Stack: Next.js + FastAPI + PostgreSQL + Redis/Celery

> Sightline is a private intelligence aggregation platform built around Automatic License

> Plate Recognition (ALPR) data. It ingests Flock-style ALPR exports, normalizes and

> deduplicates them, maps camera networks, correlates vehicle sightings into investigation

> timelines, manages watchlists and alerts, generates analyst reports, and enriches

> investigations with passive OSINT. ALPR is the core product; OSINT is a supporting module.

---

## 1. Executive Summary

Investigators working with ALPR networks (Flock Safety, Motorola Vigilant, Genetec AutoVu,

Rekor, and homegrown camera meshes) drown in raw plate-read exports: millions of timestamped

rows with inconsistent plate formatting, camera identifiers, and coordinate precision. Turning

that firehose into a defensible investigation - *who was where, when, and how they moved* -

is manual, slow, and error-prone.

Sightline is the system of record between the raw export and the analyst's conclusion. It

provides a normalized data model, a deterministic ingest/dedup pipeline, geospatial camera

mapping, plate/partial-plate search, timeline and route reconstruction, watchlist alerting,

an entity-relationship graph, evidence chain-of-custody, and exportable case reports. Every

write is attributed to a source and recorded in an immutable audit log, so findings are

traceable and the platform is defensible in a privacy-sensitive context.

**Design stance on responsibility:** because ALPR is privacy-sensitive, access control, full

audit logging, source attribution, and configurable retention are *first-class* requirements,

not afterthoughts. The platform assumes the operator already holds the data lawfully and is

accountable for its use; Sightline's job is to make that use traceable and disciplined.

## 2. Product Vision

A single-tenant-per-deployment (multi-org capable) investigation platform that an analyst can

stand up with `docker compose up`, point at a folder of Flock CSV/JSON exports, and within

minutes be searching plates, watching a camera map light up with sightings, and assembling a

timeline into a court-presentable report. The OSINT enrichment layer adds context (domains,

infrastructure, public footprint) when an investigation crosses into the digital world, but it

never overshadows the ALPR core.

Differentiators:

- **Deterministic, idempotent ingest** - re-importing the same export never double-counts.

- **Raw preservation + normalized projection** - every normalized row points back to the exact

  raw bytes it came from, for evidentiary traceability.

- **Movement intelligence** - route reconstruction, dwell detection, camera-to-camera transit

  graphs, frequency/co-travel analysis, not just a searchable table.

- **Pluggable enrichment** - a uniform enricher contract so new OSINT sources drop in without

  touching core code.

## 3. MVP Scope

**In (MVP):**

- Auth + RBAC (admin / analyst / viewer), JWT sessions, audit logging.

- Investigation CRUD; attach sightings, notes, evidence.

- ALPR ingest: CSV + JSON parsers, column auto-mapping, normalization, dedup, raw preservation.

- Cameras, camera locations, camera groups; geospatial storage; camera map API.

- Plate / partial-plate / vehicle / time-window / camera / location search.

- Sighting timeline + route reconstruction + first/last-seen + frequency analytics.

- Watchlists, watchlist entries, alert rules, watchlist-hit generation on ingest.

- Entity graph (plates↔vehicles↔sightings↔cameras↔incidents) read API.

- Evidence upload with hash + source attribution.

- Report builder (JSON/CSV/Markdown; PDF structure defined) .

- OSINT enrichment plugins: DNS, RDAP/WHOIS, cert-transparency, IP/ASN, security.txt,

  robots.txt, GitHub public metadata (the rest are stubbed against the same contract).

- Docker Compose dev environment, Alembic migrations, seed/sample data, pipeline tests.

**Delivered beyond the original MVP cutline:** PDF parser (pdfminer.six), PDF report export

(reportlab), live API connectors (JSON-REST + webhook), real-time stream ingest, S3-compatible

storage backend, OIDC/SSO login, API-key auth, evidence chain-of-custody, cascade-safe retention

enforcement (admin + scheduled). **Still deferred (true post-MVP):** multi-tenant billing,

mobile client, Mapbox vector styling, OCR for scanned (non-text-layer) PDFs.

## 4. Full Architecture

```

                         ┌──────────────────────────────────────────┐

                         │            Next.js / React (TS)            │

                         │  Dashboard · Investigations · Camera Map   │

                         │  Plate Search · Timeline · Watchlists ·    │

                         │  Entity Graph · OSINT Panel · Reports      │

                         └───────────────┬──────────────────────────┘

                                         │  HTTPS / REST + JWT

                         ┌───────────────▼──────────────────────────┐

                         │                FastAPI                     │

                         │  Auth/RBAC · Investigations · Search ·     │

                         │  Imports · Cameras · Watchlists · Graph ·  │

                         │  Reports · Enrichment · Audit middleware   │

                         └───┬───────────────┬───────────────┬───────┘

                             │               │               │

                  ┌──────────▼───┐   ┌───────▼───────┐  ┌────▼─────────┐

                  │ PostgreSQL    │   │ Redis + Celery │  │ Object store │

                  │ (PostGIS-     │   │ ingest jobs,   │  │ local FS dev │

                  │  ready, FTS,  │   │ enrichment,    │  │ S3 prod      │

                  │  pg_trgm)     │   │ report render  │  │ (evidence)   │

                  └───────────────┘   └───────┬───────┘  └──────────────┘

                                              │

                              ┌───────────────▼───────────────┐

                              │ Parser pipeline (CSV/JSON/PDF) │

                              │ Normalizer · Deduper · Analytics│

                              │ Enricher plugins (OSINT)        │

                              └────────────────────────────────┘

```

Request flow for ingest: upload → store raw blob + `imports` row → enqueue Celery job →

parser detects schema → per-row normalize → dedup against `(plate_norm, camera_id, read_ts)`

key → insert `plate_reads`/`sightings` linked to raw → evaluate watchlists → emit

`watchlist_hits` + alerts → write `audit_logs`. All synchronous APIs (search, timeline, graph)

read the normalized projection only.

## 5. Text Architecture Diagram (data lifecycle)

```

RAW EXPORT (csv/json/pdf/manual)

   │  bytes preserved verbatim

   ▼

imports(raw_blob_ref, sha256, source_id, status)          ← evidentiary anchor

   │

   ▼  parser → row dicts

normalize(): plate→UPPER/strip, ts→tz-aware UTC, geo→(lat,lon),

             direction→enum, confidence→0..1, camera_id→source-scoped

   │

   ▼  dedup key = (org, plate_norm, camera_id, read_ts_bucketed)

plate_reads(raw_read jsonb, normalized cols, import_id, dedup_hash UNIQUE)

   │           │

   │           └── upsert vehicle (by plate) , upsert camera (by source camera id)

   ▼

sightings(plate_read_id, vehicle_id, camera_id, location, ts)  ← investigation-attachable

   │

   ├── watchlist match → watchlist_hits → alerts

   ├── analytics (first/last seen, route, dwell, frequency, co-travel)

   └── entity_relationships (graph projection)

```

## 6. Database Schema

PostgreSQL. Geospatial via PostGIS when available, with a lat/lon + GiST-on-point fallback so

the schema runs on stock Postgres for the portfolio demo. Full-text via `tsvector`; partial

plate via `pg_trgm` GIN index. Times are `timestamptz` (UTC) with original tz preserved in raw.

Core tables (abridged DDL; full SQLAlchemy models in `backend/app/models/`):

```sql

-- identity & access

users(id uuid pk, email citext unique, hashed_password, full_name, role_id fk,

      is_active bool, created_at, updated_at)

roles(id, name unique, permissions jsonb)               -- admin/analyst/viewer

api_keys(id, user_id fk, name, key_hash, scopes jsonb, last_used_at, revoked_at, created_at)

audit_logs(id bigserial, actor_user_id, action, entity_type, entity_id, ip,

           metadata jsonb, created_at)  -- INDEX(entity_type, entity_id), INDEX(created_at)

-- sources & ingest

sources(id, name, system_type, organization, jurisdiction, trust_level, created_at)

imports(id, source_id fk, filename, content_type, sha256 unique, raw_blob_ref,

        row_count, status, error, created_by, created_at)  -- raw preservation

enrichment_jobs(id, investigation_id, enricher, target, status, result jsonb,

                confidence, raw_response_ref, created_at, finished_at)

-- cameras / geo

locations(id, label, address, lat double precision, lon double precision,

          geom geography(Point,4326) NULL, jurisdiction, created_at)  -- GiST(geom)

cameras(id, source_id fk, external_camera_id, name, direction_facing, location_id fk,

        camera_group_id fk, metadata jsonb, created_at)

        UNIQUE(source_id, external_camera_id)              -- source-scoped identity

camera_groups(id, name, organization, description)

-- plates / vehicles / reads / sightings

license_plates(id, plate_raw, plate_norm, state, plate_type, created_at)

               UNIQUE(plate_norm, state); GIN(plate_norm gin_trgm_ops)  -- partial search

vehicles(id, primary_plate_id fk, make, model, color, body_type, year, notes,

         possible_duplicate_of fk NULL, created_at)

plate_reads(id bigserial, import_id fk, source_id fk, camera_id fk, plate_id fk,

            plate_norm, read_ts timestamptz, direction, confidence numeric(4,3),

            lane, hotlist_hit bool, raw_read jsonb, dedup_hash bytea UNIQUE,

            incident_ref text, created_at)

            INDEX(plate_norm), INDEX(read_ts), INDEX(camera_id),

            INDEX(read_ts, camera_id), GIN(to_tsvector(raw))

sightings(id bigserial, plate_read_id fk, vehicle_id fk, camera_id fk, location_id fk,

          investigation_id fk NULL, read_ts timestamptz, confidence, created_at)

          INDEX(vehicle_id, read_ts), INDEX(investigation_id), INDEX(read_ts)

-- investigations & case material

investigations(id, title, status, priority, summary, lead_user_id, organization,

               created_at, updated_at)  -- status: open/active/closed/archived

targets(id, investigation_id fk, kind, label, plate_id fk NULL, vehicle_id fk NULL,

        osint_entity_id fk NULL, notes, created_at)

incidents(id, external_ref, title, kind, occurred_at, location_id fk, description, created_at)

notes(id, investigation_id fk, author_id, body, pinned bool, created_at)

tags(id, name unique, color); taggings(tag_id, entity_type, entity_id)  -- polymorphic

evidence_files(id, investigation_id fk, filename, content_type, size_bytes, sha256,

               blob_ref, source_id fk, description, uploaded_by, created_at)

-- watchlists & alerts

watchlists(id, name, organization, kind, active bool, created_by, created_at)

watchlist_entries(id, watchlist_id fk, plate_norm NULL, partial_plate NULL,

                  vehicle_desc jsonb NULL, reason, expires_at, created_at)

watchlist_hits(id, watchlist_entry_id fk, plate_read_id fk, sighting_id fk,

               matched_on, created_at)  -- INDEX(created_at)

alert_rules(id, watchlist_id fk NULL, condition jsonb, channel, throttle_seconds, active)

alerts(id, alert_rule_id fk, watchlist_hit_id fk, status, payload jsonb, created_at)

-- graph & osint

osint_entities(id, kind, value, attributes jsonb, first_seen, last_seen, created_at)

               UNIQUE(kind, value)

entity_relationships(id, src_type, src_id, dst_type, dst_id, relation, weight,

                     evidence_ref, confidence, created_at)

                     INDEX(src_type, src_id), INDEX(dst_type, dst_id)

-- output

reports(id, investigation_id fk, title, format, sections jsonb, rendered_ref,

        created_by, created_at)

exports(id, investigation_id fk, kind, format, filters jsonb, file_ref, created_by, created_at)

```

**Migration strategy:** Alembic, one base revision establishing the schema, `pg_trgm` +

(optional) `postgis` extensions created conditionally in the first migration. Forward-only in

dev; reversible downgrades authored for each revision. Seed data loaded via an idempotent

`seed.py` (upserts, safe to re-run).

## 7. API Route Design (REST, `/api/v1`)

```

Auth        POST   /auth/register            POST /auth/login (->JWT)   GET /auth/me

Users/RBAC  GET    /users  POST /users  PATCH /users/{id}  (admin)

            GET    /roles

Sources     GET/POST /sources

Imports     POST   /imports            (multipart upload -> enqueue)

            GET    /imports  GET /imports/{id}  (status, row_count, errors)

Cameras     GET/POST /cameras   GET /cameras/{id}   GET /cameras/map?bbox=&group=

            GET    /camera-groups  POST /camera-groups

Plates      GET    /plates/search?q=&partial=&state=

            GET    /plates/{plate_norm}             (profile + sightings)

Vehicles    GET    /vehicles/{id}  PATCH /vehicles/{id}  (merge duplicates)

Sightings   GET    /sightings?plate=&camera=&from=&to=&bbox=&investigation=

Search      POST   /search          (unified: plate/partial/vehicle/time/geo/camera/incident)

Timeline    GET    /investigations/{id}/timeline

Analytics   GET    /vehicles/{id}/movement     (route, dwell, frequency, co-travel)

            GET    /cameras/{id}/transits       (camera-to-camera graph)

Invest.     GET/POST /investigations  GET/PATCH/DELETE /investigations/{id}

            POST   /investigations/{id}/sightings   POST /{id}/notes  POST /{id}/evidence

            GET    /investigations/{id}/graph       (entity graph)

Watchlists  GET/POST /watchlists  POST /watchlists/{id}/entries

            GET    /watchlist-hits?from=&to=

Alerts      GET    /alerts  PATCH /alerts/{id}  (ack/resolve)

Enrichment  POST   /enrichment/run  {investigation_id, enricher, target}

            GET    /enrichment/jobs/{id}    GET /enrichers   (registry)

Graph       GET    /graph?root_type=&root_id=&depth=

Reports     POST   /investigations/{id}/reports   GET /reports/{id}   GET /reports/{id}/download

Exports     POST   /investigations/{id}/exports   GET /exports/{id}/download

Audit       GET    /audit?entity_type=&entity_id=&actor=&from=&to=   (admin/analyst)

Admin       GET    /admin/settings  PATCH /admin/settings   (retention, rate limits)

Health      GET    /health  GET /health/ready

```

Auth: `Authorization: Bearer <JWT>` or `X-API-Key`. RBAC enforced by a dependency that checks

`role.permissions` against a per-route required-permission. Every mutating route writes an

`audit_logs` row via middleware.

## 8. Parser / Import Pipeline

1. **Upload** (`POST /imports`): stream file to blob store, compute SHA-256, create `imports`

   row (`status=received`). If a prior import shares the SHA-256 → short-circuit (idempotent).

2. **Enqueue** Celery `process_import(import_id)`.

3. **Detect**: sniff content-type + header row; choose `CsvFlockParser` / `JsonFlockParser` /

   `PdfReportParser`. Parsers implement a common `iter_rows() -> Iterable[RawRead]`.

4. **Column mapping**: heuristic mapper handles Flock/Vigilant/Genetec header variants

   (`plate`, `Plate`, `license_plate`, `lp`, `tag`; `timestamp`, `read_time`, `captured_at`;

   `camera`, `device_id`, `cam`; `lat/lon`, `latitude/longitude`, `gps`). Unmapped columns are

   kept in `raw_read` jsonb.

5. **Normalize** (`services/normalize.py`): plate → uppercase, strip non-alphanumerics, fold

   look-alikes optionally; timestamp → tz-aware UTC (parse offset or apply source tz);

   confidence → 0..1; direction → {N,NE,E,SE,S,SW,W,NW,UNKNOWN}; coordinates validated to

   plausible ranges; camera resolved/created by `(source, external_camera_id)`.

6. **Dedup**: `dedup_hash = sha256(org | plate_norm | camera_id | read_ts_to_second)`. Unique

   constraint makes re-ingest a no-op; near-duplicate (same plate+camera within N seconds)

   collapses to one read with a `burst_count`.

7. **Persist**: insert `plate_reads` (linked to `import_id`), upsert `license_plates` +

   `vehicles`, create `sightings`.

8. **Watchlist eval**: match each new read against active `watchlist_entries`

   (exact, partial-prefix, vehicle-desc) → `watchlist_hits` → `alerts` (throttled).

9. **Finalize**: update `imports.row_count/status`, write `audit_logs`, surface errors per-row

   in `imports.error` without failing the whole batch (row-level resilience).

## 9. ALPR Analytics Design (`services/analytics/`)

- **first/last seen**: `MIN/MAX(read_ts)` per `vehicle_id`, optionally bounded by bbox/time.

- **sightings by camera / time window / near incident**: grouped counts; "near incident" uses

  geospatial distance + time delta around an `incident`.

- **route reconstruction**: order a vehicle's sightings by `read_ts`, emit ordered camera→camera

  legs with elapsed time, distance (haversine), and implied speed; flag implausible legs

  (speed > threshold ⇒ possible plate misread or cloned plate).

- **dwell detection**: consecutive reads at the same camera/location cluster within a window ⇒

  dwell episodes (arrive_ts, depart_ts, duration).

- **camera-to-camera movement / transit graph**: weighted directed edges between cameras by

  observed transitions; powers "where does this vehicle usually go next."

- **common locations / frequency**: histogram of cameras/locations per vehicle; identifies home

  /work anchor points by time-of-day clustering.

- **co-travel**: vehicles repeatedly seen at the same camera within ±N seconds across multiple

  days ⇒ candidate associates (with a false-positive guard on high-traffic cameras).

- **possible duplicate vehicles**: same plate_norm with divergent vehicle descriptions, or

  impossible-travel ⇒ flag `possible_duplicate_of`.

- **confidence scoring**: per-read confidence carried through; aggregate finding confidence =

  function of read confidences, corroborating cameras, and source trust_level.

## 10. Watchlist & Alert Design

- **Watchlist entry kinds**: exact plate, partial/prefix plate, vehicle description (make/color

  /body), and OSINT entity. Entries can expire (`expires_at`).

- **Evaluation**: runs inline during ingest (new reads) and on-demand (backfill against history).

- **Alert rules**: `condition` jsonb (e.g. `{first_seen_in_area: true}`,

  `{camera_group: "downtown"}`, `{min_confidence: 0.85}`) bound to a watchlist; `throttle_seconds`

  prevents alert storms; `channel` = in-app (MVP), webhook/email (extensible).

- **Hit → alert**: a `watchlist_hit` records what matched (`matched_on`); an `alert` is emitted

  if a rule passes throttling. Alerts have ack/resolve lifecycle, visible in the Alert Center.

## 11. Frontend Page Plan (Next.js App Router)

| Route | Page | Key components |

|---|---|---|

| `/` | Dashboard | KPI cards, recent hits, ingest status, alert feed |

| `/investigations` | Investigation list | table, filters, status chips |

| `/investigations/[id]` | Investigation detail | tabs: overview, sightings, timeline, graph, evidence, notes, report |

| `/import` | ALPR import | drag-drop upload, source picker, mapping preview, job status |

| `/map` | Camera map | Leaflet, camera markers, sighting heat, bbox query |

| `/cameras/[id]` | Camera detail | reads over time, transit graph, location |

| `/search` | Plate search | unified search form, partial-plate, results table |

| `/vehicles/[id]` | Vehicle profile | reads, route map, frequency, co-travel |

| `/investigations/[id]/timeline` | Timeline | vertical time-ordered sightings + map sync |

| `/watchlists` | Watchlist manager | lists, entries, expiries |

| `/alerts` | Alert center | live hits, ack/resolve |

| `/evidence/[id]` | Evidence viewer | file preview, hash, chain-of-custody |

| `/graph` | Entity graph | Cytoscape/React Flow node-link |

| `/enrichment` | OSINT panel | run enricher, view normalized entities + evidence |

| `/reports` | Report builder | section toggles, preview, export |

| `/audit` | Audit log viewer | filterable immutable log |

| `/admin` | Admin/settings | users, roles, retention, API keys, rate limits |

Shared: API client (`lib/api.ts`) with JWT handling, typed DTOs mirroring backend schemas,

auth guard, role-gated nav.

## 12. File Tree

```

Sightline/

├── docker-compose.yml          # postgres, redis, api, worker, frontend

├── .env.example

├── README.md

├── docs/DESIGN.md              # this document

├── backend/

│   ├── pyproject.toml / requirements.txt

│   ├── alembic.ini · alembic/env.py · alembic/versions/0001_initial.py

│   ├── seed.py                 # idempotent sample-data loader

│   ├── app/

│   │   ├── main.py             # FastAPI app, middleware, router mount

│   │   ├── core/{config,security,permissions,logging}.py

│   │   ├── db/{base,session}.py

│   │   ├── models/*.py         # SQLAlchemy: all core entities

│   │   ├── schemas/*.py        # Pydantic DTOs

│   │   ├── api/deps.py · api/routes/*.py   # routers per domain

│   │   ├── services/

│   │   │   ├── normalize.py · dedup.py · ingest.py

│   │   │   ├── parsers/{base,csv_flock,json_flock,pdf_report}.py

│   │   │   ├── analytics/{movement,frequency,cotravel,dwell}.py

│   │   │   ├── watchlist.py · reporting.py

│   │   │   └── enrichment/{base,registry,dns,rdap,crtsh,ipasn,securitytxt,robots,github}.py

│   │   └── workers/{celery_app,tasks}.py

│   ├── sample_data/flock_sample.csv · flock_sample.json

│   └── tests/test_pipeline.py · test_analytics.py

└── frontend/

    ├── package.json · next.config.js · tsconfig.json

    └── src/{app/*, components/*, lib/api.ts}

```

## 13. Implementation Plan

1. **Foundation** - config, DB session, base models, security/JWT, RBAC permissions. *(done in MVP)*

2. **Schema & migration** - all SQLAlchemy models + Alembic base revision + seed. *(done)*

3. **Ingest core** - parsers, normalize, dedup, ingest service, import API, Celery task. *(done)*

4. **Search & analytics** - search API, timeline, movement/frequency/co-travel/dwell. *(done)*

5. **Watchlists & alerts** - entry matching during ingest, hits, alert rules. *(done)*

6. **Graph & enrichment** - entity-relationship projection, enricher contract + plugins. *(done)*

7. **Reports & exports** - section assembly, JSON/CSV/Markdown render, PDF structure. *(done)*

8. **Frontend** - API client, auth, dashboard, search, map, timeline, investigation detail. *(skeleton)*

9. **Hardening** - rate limiting, input validation, retention jobs, backups, tests. *(partial)*

MVP cutline = steps 1-7 functional and tested end-to-end on sample data; step 8 a navigable

skeleton wired to the live API; step 9 the documented-and-partially-implemented hardening floor.

## 14. Implementation (code)

See the repository. The backend ALPR pipeline (ingest → normalize → dedup → search →

analytics → watchlist) is implemented and covered by `backend/tests/`. Run the demo:

```bash

cp .env.example .env

docker compose up -d postgres redis

cd backend && pip install -r requirements.txt

alembic upgrade head && python seed.py

uvicorn app.main:app --reload         # API at http://localhost:8000/docs

# or full stack:

docker compose up --build

```