Client/Loader Pipeline Architecture

Abstract

The Prismatic DD Client/Loader Pipeline is a two-phase entity loading architecture that replaces the legacy bulk processing system with a PostgreSQL-backed, diff-aware pipeline. Sources self-register via Elixir metaprogramming (behaviour + macro + @after_compile hook), enabling zero-configuration addition of new data sources. The Client phase fetches raw data from OSINT adapters into staging tables, while the Loader phase normalizes, deduplicates, and persists entities and relationships into fortress PostgreSQL tables. A GenServer-based Scheduler manages periodic refresh cycles with configurable intervals per source group.

Introduction

The Entity Loading Challenge

Due diligence platforms must ingest entities from dozens of heterogeneous data sources -- government registries, business databases, sanctions lists, and public records -- each with different data formats, update frequencies, and reliability characteristics. The challenge is threefold:

1. Source Heterogeneity: Each source returns data in a different format. Forbes CZ provides ranked wealth lists, the Czech Parliament API provides member directories, and sanctions databases provide structured watchlists. A unified loading pipeline must normalize all of these into a consistent entity model.

2. Change Detection: Re-fetching 50,000+ entities daily is wasteful if only a fraction have changed. Content-hash-based diff detection enables the pipeline to skip unchanged records, reducing database writes and processing time by orders of magnitude.

3. Source Registration: Adding new data sources should require minimal boilerplate. The self-registering source pattern -- borrowed from the platform's OSINT tool registry -- enables new sources to appear in the pipeline and UI automatically upon compilation.

Architecture Overview

flowchart TD
    Sources["Source Modules<br/><i>self-registering via @after_compile</i>"]
    Reg["SourceRegistry<br/><i>ETS GenServer</i>"]
    Client["Client<br/><i>fetch</i>"]
    Sched["Scheduler<br/><i>periodic</i>"]
    Stage["dd_fetch_records<br/><i>PostgreSQL staging</i>"]
    Loader["Loader<br/><i>normalize + persist</i>"]
    Entities["dd_entities"]
    Rels["dd_relationships"]
    Attrs["dd_entity_attributes"]
    Fortress["PostgreSQL fortress tables"]

    Sources --> Reg
    Reg --> Client
    Reg --> Sched
    Client --> Stage
    Sched --> Loader
    Stage --> Loader
    Loader --> Entities
    Loader --> Rels
    Loader --> Attrs
    Entities --- Fortress
    Rels --- Fortress
    Attrs --- Fortress

    classDef source fill:#0f172a,stroke:#3b82f6,color:#fff
    classDef reg fill:#1e293b,stroke:#a855f7,color:#fff
    classDef worker fill:#1e293b,stroke:#10b981,color:#fff
    classDef stage fill:#0f172a,stroke:#f59e0b,color:#fff
    classDef store fill:#0f172a,stroke:#475569,color:#e2e8f0
    class Sources source
    class Reg reg
    class Client,Sched,Loader worker
    class Stage stage
    class Entities,Rels,Attrs,Fortress store

Self-Registering Source Pattern

The Source Behaviour

Every DD data source implements the PrismaticDd.Source behaviour, which defines four callbacks:

Metaprogramming Registration

The use PrismaticDd.Source macro injects three things:

1. Behaviour declaration: @behaviour PrismaticDd.Source
2. Import: register_source/1 macro
3. Default implementation: extract_relationships/2 returns [] (overridable)

The register_source/1 macro stores the configuration as a module attribute and sets up an @after_compile hook that fires during compilation:

defmacro register_source(config) do
  quote do
    @source_config unquote(config)
    @impl PrismaticDd.Source
    def get_source_config, do: @source_config
    @after_compile PrismaticDd.Source
  end
end

The __after_compile__/2 callback extracts the @source_config attribute from the compiled BEAM bytecode and registers it in the SourceRegistry ETS table:

def __after_compile__(_env, bytecode) do
  case :beam_lib.chunks(bytecode, [:attributes]) do
    {:ok, {module, [{:attributes, attributes}]}} ->
      case Keyword.get(attributes, :source_config) do
        [config] when is_map(config) ->
          PrismaticDd.SourceRegistry.register_source(module, config)
        _ -> :ok
      end
    _ -> :ok
  end
end

This pattern mirrors the OSINT tool self-registration used elsewhere in the platform, ensuring consistency across the codebase.

Source Implementation Example

defmodule PrismaticDd.Sources.ForbesCz do
  use PrismaticDd.Source

  register_source(%{
    slug: "forbes-cz",
    name: "Forbes Czech Republic",
    description: "Top 100 wealthiest Czechs from Forbes CZ annual ranking",
    group: :forbes,
    entity_type: :person,
    estimated_count: 100,
    refresh_interval_hours: 168,
    adapter_module: PrismaticOsintSources.Adapters.Czech.ForbesCz,
    tags: ["czech", "wealth", "business"]
  })

  @impl true
  def fetch(opts) do
    PrismaticOsintSources.Adapters.Czech.ForbesCz.list_richest(opts)
  end

  @impl true
  def normalize(raw_data) do
    %{
      name: raw_data["name"],
      type: "person",
      source_id: raw_data["rank"] |> to_string(),
      identifiers: %{},
      attributes: %{
        "net_worth_czk" => raw_data["net_worth"],
        "rank" => raw_data["rank"],
        "industry" => raw_data["industry"]
      }
    }
  end
end

SourceRegistry GenServer

The PrismaticDd.SourceRegistry is an ETS-backed GenServer providing sub-millisecond lookups:

Client Phase (Fetch)

The PrismaticDd.Client module implements the fetch phase:

1. Source resolution: Queries SourceRegistry for all sources in the target group
2. Adapter dispatch: Calls source.fetch(opts) on each source module
3. Content hashing: Computes MD5 hash of each raw record for diff detection
4. Staging insert: Bulk inserts raw data into dd_fetch_records with conflict skip
5. Run tracking: Creates a LoadRun record with fetch statistics
6. PubSub broadcast: Emits progress events on "dd:pipeline" topic

Key features:

• Dry run mode: Preview without side effects (Client.fetch_group(:forbes, dry_run: true))
• Fetch versioning: Each fetch creates a new version for temporal tracking
• Conflict resolution: ON CONFLICT DO NOTHING on (source_group, source_id, fetch_version) prevents duplicates

Loader Phase (Persist)

The PrismaticDd.Loader module implements the load phase:

1. Pending records: Reads dd_fetch_records with status = "pending" for the group
2. Source resolution: Resolves the source module for normalization
3. Diff detection: Compares content_hash of fetch record vs existing entity
4. Normalization: Calls source.normalize(raw_data) to produce entity attributes
5. Entity upsert: Inserts or updates dd_entities based on source_group + source_id
6. Relationship extraction: Calls source.extract_relationships/2 and upserts relationships
7. Status update: Marks fetch records as loaded, skipped, or failed
8. Run tracking: Creates a LoadRun record with load statistics

Key features:

• Skip unchanged: Records with matching content hashes are skipped (status: "skipped")
• Force mode: Loader.load_group(:parliament, force: true) ignores content hash diff
• Relationship loading: Extracts and persists entity-to-entity relationships

PostgreSQL Fortress Tables

The pipeline persists data into five PostgreSQL tables with UUID primary keys:

All tables use timestamps() with UTC precision and JSONB columns for flexible attribute storage.

Scheduler

The PrismaticDd.Scheduler GenServer manages periodic refresh cycles:

The Scheduler uses Process.send_after/3 for timer management (no external dependencies) and delegates pipeline execution to Task.Supervisor workers. Active run deduplication prevents concurrent runs of the same group.

Scheduler API

PrismaticDd.Scheduler.status()   # Current schedule state
PrismaticDd.Scheduler.pause()    # Pause all scheduled runs
PrismaticDd.Scheduler.resume()   # Resume scheduling
PrismaticDd.Scheduler.trigger(:forbes)  # Immediate fetch+load

LiveView Dashboard

The Pipeline Dashboard at /hub/dd/pipeline provides real-time monitoring:

• Source group grid: 10 groups displayed as color-coded cards with entity counts and status
• Pipeline controls: Per-group fetch/load/full pipeline triggers, plus "Run All" for full sweep
• Scheduler controls: Pause/resume toggle, next run times, active run indicators
• View modes: Grid view (cards) and table view (dense data)
• Real-time updates: PubSub subscription to "dd:pipeline" topic for live progress
• Selected group detail: Click a group to see source modules, entity counts, and last run stats

Registered Sources

Additional sources (sanctions, PEP, top firms, ARES sweep, ISIR, court parties) are planned for future registration.

References

• Platform Architecture -- Full technical architecture of the DD subsystem
• Entity Management -- Entity data model and operations
• Czech Registry Integration -- 30+ Czech registry adapters
• OSINT Integration -- 122 OSINT source framework
• OSINT Sources -- Complete source catalog
• Glossary: Elixir Macro -- Metaprogramming and compile-time code generation
• Glossary: GenServer -- OTP process pattern
• Glossary: ETS -- Erlang Term Storage

Connect & Contribute

Created by Tomas Korcak (korczis) | Open Source under GHL

• GitHub | GitLab | LinkedIn | Contact
• Developer Portal | Architecture | Meet the Creator