Sync Worker Microservice

Relevant source files

The following files were used as context for generating this wiki page:

• apps/workers/sync-worker/src/main.rs
• apps/workers/sync-worker/src/sync.rs
• apps/workers/sync-worker/src/sources/ckan.rs
• apps/workers/sync-worker/src/resource_embed.rs
• apps/workers/sync-worker/README.md
• apps/workers/sync-worker/Cargo.toml

The Sync Worker is a specialized Rust-based microservice within the Poneglyph ecosystem designed to synchronize humanitarian datasets from external repositories (such as CKAN-based platforms) into the project's central database. Its primary objective is to maintain an up-to-date repository of field reports and documents for NGO workers and researchers.

Sources: apps/workers/sync-worker/README.md:1-5, apps/server/src/agents/prompts/system.ts:10-15

The service performs metadata extraction, manages dataset versioning by archiving missing entries, and generates vector embeddings for both metadata and file content (PDF, CSV, JSON, XLSX) to enable semantic search capabilities. It is architected to run in multiple environments, supporting local development, Cloud Run, or AWS Lambda.

Sources: apps/workers/sync-worker/README.md:7-15, apps/workers/sync-worker/src/sync.rs:80-95

Architecture and Deployment Modes

The microservice is built using the Tokio runtime and provides a dual-entry point system to allow flexible deployment. It can act as a long-running HTTP server or an ephemeral serverless function.

Deployment Runtimes

The service detects its environment at startup to determine the execution path:

• AWS Lambda Mode: Activated if AWS_LAMBDA_FUNCTION_NAME is present. It uses lambda_http to handle requests from API Gateway or EventBridge.
• Local/Cloud Run Mode: Starts a TCP listener using tokio::net::TcpListener on a configured port (defaulting to 8080) to handle standard HTTP traffic.

Sources: apps/workers/sync-worker/src/main.rs:32-41, apps/workers/sync-worker/README.md:7-12

Component Relationship

The following diagram illustrates how the Sync Worker interacts with external sources and internal storage:

Synchronization Logic

The core logic resides in the run_sync function, which follows a strictly defined workflow to ensure data integrity and minimize redundant processing.

Sync Workflow Steps

1. Source Identification: Resolves the internal source_id based on the request.
2. External Fetching: Queries external APIs (e.g., CKAN) using a search query (default: "Survey Data").
3. Diffing: Compares fetched external IDs against existing IDs in the database to identify truly new datasets.
4. Logging: Creates a sync_log entry to track the operation's progress.
5. Upsert & Tagging: Persists dataset records and links associated tags.
6. Archival: Identifies datasets previously synced from this source that are no longer present in the external fetch and marks them as archived.

Sources: apps/workers/sync-worker/src/sync.rs:10-60, apps/workers/sync-worker/src/sync.rs:115-125

Implementation Detail: Sync Logic

// apps/workers/sync-worker/src/sync.rs:35-45
let datasets = sources::fetch_all(&req.source, query).await?;
let all_external_ids: Vec<String> = datasets.iter().map(|d| d.external_id.clone()).collect();
let existing_external_ids =
    db::get_existing_external_ids(pool, source_id, &all_external_ids).await?;

let mut new_datasets: Vec<&SourceDataset> = Vec::with_capacity(datasets.len());
for ds in &datasets {
    if !existing_external_ids.contains(&ds.external_id) {
        new_datasets.push(ds);
    }
}

Embedding System

The Sync Worker utilizes a "best-effort" embedding strategy to facilitate AI-powered humanitarian intelligence analysis.

Embedding Multi-modal Data

The service generates embeddings in two stages:

1. Metadata Embedding: A text representation is built from the title, description, publisher, and tags.
2. Resource Embedding: If the dataset contains supported file types, the worker downloads them and extracts content.
   • PDF: Uploaded to Gemini File API for processing.
   • CSV/JSON: Processed as plain text.
   • XLSX/XLS: Parsed using the calamine crate to extract cell data.

Sources: apps/workers/sync-worker/src/resource_embed.rs:60-90, apps/workers/sync-worker/src/sources/ckan.rs:145-160

Resource Processing Limits

To prevent unbounded resource consumption or API costs, the worker enforces several limits:

Limit Description	Config Key / Constraint
Max Resources per Dataset	config::CONFIG.max_resources_per_dataset
Max Download Size	config::CONFIG.max_download_bytes
Max Text Characters	config::CONFIG.max_text_chars
Supported Formats	PDF, CSV, JSON, XLSX, XLS

Sources: apps/workers/sync-worker/src/resource_embed.rs:50-58, apps/workers/sync-worker/README.md:15-18

Embedding Flow Sequence

The sequence below describes the interaction between the worker and the Gemini API during file processing:

API Interface

The worker exposes a minimal HTTP interface for triggering synchronization and health monitoring.

Endpoints

Method	Path	Description	Payload
POST	/sync	Triggers a synchronization run for a specific source.	SyncRequest (JSON)
GET	/health	Returns {"status": "healthy"}.	N/A

Sources: apps/workers/sync-worker/src/main.rs:75-105

Data Structures

SyncRequest

• source: String (The name of the source, e.g., "opencity")
• source_id: Optional UUID
• query: Optional String (Search term for external API)

Sources: apps/workers/sync-worker/src/sync.rs:10-15, apps/workers/sync-worker/src/main.rs:80-85

Conclusion

The Sync Worker Microservice is a critical infrastructure component that automates the ingestion of external data into the Poneglyph platform. By combining robust synchronization logic with deep resource embedding using Gemini, it ensures that humanitarian analysts have access to a searchable, semantically-indexed repository of global survey data. Sources: apps/workers/sync-worker/README.md, apps/workers/sync-worker/src/sync.rs