How Macro Implements Contact and CRM Synchronization Across Email and Messages

Macro synchronizes contacts and CRM data across email and messaging by centralizing contact records in the crm_contact table and using the sync_contacts utility in the email service to link incoming email addresses with canonical CRM entries.

The macro-inc/macro repository implements a unified contact management system that bridges email ingestion pipelines with CRM storage and messaging interfaces. By leveraging background pub/sub workers and shared database tables, the platform ensures that contact information remains consistent whether it originates from Gmail synchronization or manual CRM entry.

Centralized Contact Storage with crm_contact

Database Schema and Migrations

The foundation of contact and CRM synchronization rests on the crm_contact table, defined in migration files located at crates/macro_db_client/migrations/20260520150000_crm_contacts_name.*. These migrations establish the canonical storage for contact records across the entire application.

The crm_contact Table Structure

The table stores essential contact metadata including unique identifiers, display names, email addresses, data sources, and visibility flags. This structure serves as the single source of truth for both email processing and CRM operations.

-- Migration creating the canonical contacts table
CREATE TABLE "crm_contact" (
    "id" BIGINT PRIMARY KEY,
    "name" TEXT,
    "email" TEXT UNIQUE,
    "source" TEXT,
    "hidden" BOOLEAN DEFAULT FALSE,
    "created_at" TIMESTAMP NOT NULL DEFAULT now(),
    "updated_at" TIMESTAMP NOT NULL DEFAULT now()
);

Email Service Contact Synchronization

The sync_contacts Utility

Located in services/email_service/src/util/sync_contacts.rs, the sync_contacts function manages the core synchronization logic. This utility accepts a list of email addresses extracted from incoming messages and ensures each address has a corresponding entry in the crm_contact table.

// Called after a new email thread is stored
use crate::util::sync_contacts::sync_contacts;

pub async fn process_new_thread<B: MacroEventBroker>(
    broker: B, 
    thread_id: i64
) -> Result<()> {
    // Pull email addresses from the thread
    let addresses = fetch_thread_addresses(thread_id).await?;
    
    // Ensure each address has a CRM contact entry
    sync_contacts(broker, addresses).await?;
    Ok(())
}

Back-fill Workers for CRM Consistency

The system maintains contact accuracy through scheduled back-fill jobs that process historical data:

• populate_crm_contact.rs: Located at services/email_service/src/pubsub/backfill/populate_crm_contact.rs, this worker reads Gmail contacts via the email_api_client, invokes sync_contacts, and persists records to the database.
• seed_sent_contact.rs: Processes outbound message contacts to ensure sent email addresses are tracked.
• depopulate_crm_contact.rs: Handles cleanup and removal of stale contact associations.

Linking Messages to Contacts

Message-Level Contact Association

When the system stores new email messages, the email_db_client crate extracts email addresses and creates associations through the upsert_message_contacts function in crates/email_db_client/src/contacts/upsert_message.rs. This function populates the message_contacts join table, linking specific messages to their canonical crm_contact records.

Real-Time Synchronization Flow

The synchronization pipeline follows a clear path: Gmail data enters through the email service, triggers sync_contacts to update the crm_contact table, and subsequently updates the message_contacts join table. This ensures that the messaging layer can immediately resolve contact identity queries for UI components.

CRM API and Frontend Integration

CRM Service Endpoints

The CRM API exposes standardized endpoints for contact management through files in crates/crm/src/inbound/axum_router/*.rs. These routes implement operations such as create_contact, set_contact_name, and list_company_contacts, all reading from and writing to the same crm_contact table that the email service maintains.

Frontend Data Fetching

The web application consumes contact data through generated SDK schemas and query utilities:

// Frontend contact list retrieval
import { getContacts } from '@/lib/queries/crm/contacts';

export async function useContactList() {
  const { data, error } = await getContacts();
  return { contacts: data ?? [], error };
}

TypeScript definitions in apps/web/src/lib/service-clients/service-email/generated/schemas/contact.ts ensure type safety across the API boundary, while apps/web/src/lib/queries/crm/contacts.ts provides the query interface for React components.

Summary

• Centralized Storage: The crm_contact table in crates/macro_db_client/migrations/ serves as the single source of truth for all contact information.
• Synchronization Engine: The sync_contacts function in services/email_service/src/util/sync_contacts.rs bridges Gmail data with CRM records.
• Background Processing: Pub/sub back-fill workers like populate_crm_contact.rs ensure data consistency across historical email archives.
• Message Association: The message_contacts join table links individual messages to canonical contacts via crates/email_db_client/src/contacts/upsert_message.rs.
• Unified API: CRM endpoints in crates/crm/src/inbound/axum_router/ and frontend queries in apps/web/src/lib/queries/crm/contacts.ts provide consistent access to synchronized data.

Frequently Asked Questions

How does Macro handle duplicate contacts from different email sources?

The sync_contacts utility checks for existing records using the unique email field in the crm_contact table before creating new entries. When duplicates are detected, the system updates the existing record rather than creating a new one, ensuring that a single email address maps to exactly one canonical contact regardless of how many threads or sources reference it.

What triggers the contact synchronization process in the email service?

Contact synchronization triggers in two scenarios: immediately when new email threads are processed through process_new_thread, and periodically via background pub/sub workers such as populate_crm_contact.rs. The real-time trigger ensures new conversations reflect current contact data, while back-fill jobs maintain historical accuracy.

How do message threads link to CRM contacts in the database?

The system creates entries in the message_contacts join table through the upsert_message_contacts function. This table establishes many-to-many relationships between messages and the crm_contact table, allowing the messaging interface to display enriched contact information for any given thread.

Can the CRM contact data be accessed via API for third-party integrations?

Yes, the CRM service exposes REST endpoints through crates/crm/src/inbound/axum_router/*.rs that support standard CRUD operations on contacts. These endpoints interact directly with the crm_contact table, making synchronized contact data available to authorized external integrations and internal frontend components alike.

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