Event Processor

Overview

The tg-event-processor is a Go service that handles all disaster event ingestion into TerraGuard. It replaces a previous Lambda+SQS+DynamoDB architecture with a single deterministic binary that polls authoritative data sources, normalizes events into a common format, deduplicates them, correlates them with existing records using PostGIS spatial queries, and performs atomic upserts into PostgreSQL.

The service runs on port 5606 and compiles to a ~16MB static binary.

Architecture

Key Features

Adapter Pattern

Every data source implements the SourceAdapter interface:

type SourceAdapter interface {
    Name() string
    Poll(ctx context.Context) ([]model.RawEvent, error)
    Normalize(raw model.RawEvent) (*model.NormalizedEvent, error)
    ResolveIDs(raw model.RawEvent) []string
    Config() AdapterConfig
}

Adding a new source requires implementing this interface and registering the adapter in cmd/server/main.go. Zero pipeline code changes needed.

Deterministic Pipeline

Every event passes through the same six-stage pipeline:

1. ID Resolution -- Resolves all known identifiers for an event. USGS earthquakes can have multiple IDs from different seismic networks (e.g., us7000abc, nn00001). The event_id_aliases table tracks these relationships.

2. Hash Dedup -- Computes an MD5 hash of the raw event JSON. If the hash matches an existing ingestion record, the event is skipped (no changes since last poll).

3. Staleness Check -- Compares the source's lastUpdated timestamp against the stored value. Skips events that have not been updated since last ingestion.

4. PostGIS Correlation -- Uses spatial and temporal queries to find matching DisasterEvent records. Thresholds are configurable per event type in config/correlation.yaml.

5. Atomic Upsert -- Within a single database transaction, either creates a new DisasterEvent or updates an existing one, records the EventIngestion, and stores ID aliases.

6. Webhook Notification -- POSTs to the Backend API (/api/v1/internal/event-processed) to trigger the enrichment pipeline via Inngest.

YAML-Configurable Correlation

Correlation thresholds are defined in config/correlation.yaml with no code changes required:

USGS ID Aliasing

USGS earthquakes present a unique challenge: the same physical earthquake can have different IDs from different seismic networks. The USGS ids property contains a comma-separated list like ",us7000abc,nn00001,ci12345,". The adapter's ResolveIDs() method parses all aliases, and the pipeline stores them in the event_id_aliases table so that future updates from any network ID are correctly correlated to the same disaster event.

Testing Endpoints

The service exposes /test routes for manual event injection and data file processing, useful for replaying historical events or testing adapter changes without waiting for a poll cycle. Adapter names in test endpoint paths are case-insensitive (e.g., gdacs, GDACS, and Gdacs all work).

Available test endpoints:

Configuration

API Endpoints

Directory Structure

tg-event-processor/
├── cmd/
│   └── server/
│       └── main.go             # Entry point: config, DB, registry, scheduler, HTTP
├── internal/
│   ├── adapter/
│   │   ├── adapter.go          # SourceAdapter interface + helpers
│   │   ├── registry.go         # Adapter registry (register, get, list)
│   │   ├── gdacs.go            # GDACS GeoJSON adapter
│   │   ├── usgs.go             # USGS Earthquake adapter
│   │   └── nhc.go              # NHC ActiveStorms adapter
│   ├── api/
│   │   ├── handler.go          # HTTP handlers (health, adapters, metrics, poll)
│   │   ├── testing.go          # Test/simulation endpoints
│   │   └── middleware.go       # Auth, CORS, logging, recovery
│   ├── config/
│   │   └── config.go           # Environment configuration loading
│   ├── db/
│   │   └── db.go               # PostgreSQL connection pool + migrations
│   ├── model/
│   │   ├── event.go            # RawEvent, NormalizedEvent, enums
│   │   └── result.go           # ProcessResult
│   ├── pipeline/
│   │   ├── pipeline.go         # Core pipeline orchestration
│   │   ├── alias.go            # ID alias resolution
│   │   ├── dedup.go            # Hash-based deduplication
│   │   ├── staleness.go        # Staleness check
│   │   ├── correlation.go      # PostGIS spatial/temporal matching
│   │   ├── upsert.go           # Atomic upsert logic
│   │   └── webhook.go          # Backend webhook notifier
│   └── scheduler/
│       └── scheduler.go        # Per-adapter poll scheduling
├── config/
│   ├── correlation.yaml        # Correlation thresholds per event type
│   └── scoring.yaml            # Event scoring configuration
├── migrations/                 # SQL migration files
├── Makefile
└── go.mod

Running

# Build the binary
make build

# Build and run
make run

# Run tests
make test

# Run unit tests only (skip integration)
make test-short