CRM Conventions

Contact, Organization, and Opportunity Management on the Knowledge Graph

1. Relationship to the Core Schema

This document defines conventions, not new tables. CRM functionality is implemented on the six core tables (nodes, edges, events, event_participants, assertions, artifacts). The CRM layer consists of:

Agreed-upon values for node_type, edge_type, event_type, and assertion_type
Human-readable code conventions
Materialized views that project the graph into CRM-shaped read models
Convenience functions for common write patterns

A CRM contact and a knowledge graph person are the same node. There is no synchronization because there is no separation.

2. Entity Conventions

2.1 Node Types

`node_type`	Purpose	Required Properties	Optional Properties
`person`	Individual contacts, leads, team members	`first_name`, `last_name`	`email`, `phone`, `address`, `title`, `source`
`org`	Companies, agencies, vendors, clients	`name`	`industry`, `website`, `address`, `phone`
`opportunity`	Deals, acquisitions, sales targets	`name`, `code`	`description`, `estimated_value`, `region`
`pipeline`	A named workflow/sales process	`name`, `stages`	`description`, `default_stage`
`territory`	Geographic or logical grouping	`name`	`description`
`campaign`	Marketing or outreach initiative	`name`, `campaign_type`	`start_date`, `end_date`, `budget`

2.2 Edge Types

`edge_type`	Source -> Target	Properties
`employs`	org -> person	`title`, `department`, `start_date`
`affiliated_with`	person -> org	`relationship`, `notes`
`primary_contact`	opportunity -> person	`role`
`secondary_contact`	opportunity -> person	`role`, `notes`
`pipeline_member`	opportunity -> pipeline	`entered_at`
`territory_member`	opportunity -> territory	—
`campaign_target`	campaign -> person	`channel`, `sent_at`, `responded_at`
`assigned_to`	opportunity -> person	`role` (`'owner'`, `'backup'`)
`targets`	opportunity -> (any node)	`priority`, `notes`
`referral`	person -> person	`context`, `date`

Temporal edges (employs, assigned_to, affiliated_with) use effective_from and effective_to. When a person leaves a company, the edge gets an effective_to rather than being deleted.

2.3 Assertion Types

Use assertions for data that changes over time and where history matters. Static facts go in node properties.

`assertion_type`	Subject	Claim Shape
`deal_stage`	opportunity	`{stage, pipeline, moved_from, reason}`
`deal_value`	opportunity	`{amount, currency, basis}`
`sentiment`	person	`{stance, confidence, notes}`
`contact_info`	person	`{phone, email, address, preferred_contact}`
`lead_score`	person or opp	`{score, model, factors}`
`win_probability`	opportunity	`{probability, basis}`
`do_not_contact`	person	`{reason, set_by, expires}`

2.4 Event Types

`event_type`	Purpose	Key Properties
`email`	Email sent or received	`subject`, `direction`, `body_ref`
`phone_call`	Phone call logged	`duration`, `direction`, `outcome`
`meeting`	In-person or virtual	`location`, `duration`, `outcome`
`note`	Free-form note	`body`, `category`
`offer_sent`	Formal offer dispatched	`amount`, `terms`, `document_ref`
`stage_change`	Pipeline stage transition	`from_stage`, `to_stage`, `reason`

3. Codes and Identifiers

3.1 Format

{PREFIX}-{YYMM}-{SEQ}

Entity	Prefix	Example
Opportunity	`OPP`	`OPP-2403-0042`
Person	`CON`	`CON-2401-0156`
Organization	`ORG`	`ORG-2312-0003`
Campaign	`CMP`	`CMP-2406-0001`
Territory	`TER`	`TER-2401-0001`

Codes are generated by generate_crm_code() (see Functions) and stored in both properties->>'code' and external_id with external_source = 'internal'.

3.2 Codes in Practice

-- Exact lookup
SELECT * FROM nodes WHERE properties @> '{"code": "OPP-2403-0042"}';

-- Prefix search (all March 2024 opportunities)
SELECT * FROM nodes
WHERE node_type = 'opportunity'
    AND properties->>'code' LIKE 'OPP-2403-%';

4. Pipeline and Deal Management

4.1 Pipeline Definition

A pipeline is a node with an ordered list of stages. Multiple pipelines can coexist.

INSERT INTO nodes (node_type, label, properties) VALUES (
    'pipeline', 'Sales Pipeline',
    '{
        "name": "Sales Pipeline",
        "code": "PL-SALES",
        "stages": [
            {"key": "prospecting", "label": "Prospecting", "order": 1},
            {"key": "qualification", "label": "Qualification", "order": 2},
            {"key": "proposal", "label": "Proposal", "order": 3},
            {"key": "negotiation", "label": "Negotiation", "order": 4},
            {"key": "closed_won", "label": "Closed Won", "order": 5, "terminal": true},
            {"key": "closed_lost", "label": "Closed Lost", "order": 6, "terminal": true}
        ],
        "default_stage": "prospecting"
    }'
);

4.2 Stage Transitions

Moving through stages uses assertion supersession, preserving full history:

SELECT advance_deal_stage('<opp_uuid>', 'negotiation', 'Proposal accepted, entering terms discussion');

5. Convenience Functions

5.1 Create Opportunity

CREATE FUNCTION create_opportunity(
    p_name text,
    p_pipeline_code text,
    p_assigned_to_id uuid,
    p_properties jsonb DEFAULT '{}',
    p_teams text[] DEFAULT '{}'
) RETURNS uuid AS $$
DECLARE
    v_opp_id uuid;
    v_code text;
    v_pipeline_id uuid;
    v_default_stage text;
BEGIN
    v_code := generate_crm_code('OPP');

    SELECT id, properties->>'default_stage'
    INTO v_pipeline_id, v_default_stage
    FROM nodes
    WHERE node_type = 'pipeline' AND properties->>'code' = p_pipeline_code;

    INSERT INTO nodes (node_type, label, properties, attrs, external_id, external_source)
    VALUES (
        'opportunity', p_name,
        p_properties || jsonb_build_object('name', p_name, 'code', v_code),
        jsonb_build_object('teams', to_jsonb(p_teams)),
        v_code, 'internal'
    )
    RETURNING id INTO v_opp_id;

    INSERT INTO assertions (assertion_type, subject_node_id, claim, confidence)
    VALUES ('deal_stage', v_opp_id,
        jsonb_build_object('stage', v_default_stage, 'pipeline', p_pipeline_code), 1.0);

    INSERT INTO edges (edge_type, source_id, target_id, properties)
    VALUES ('pipeline_member', v_opp_id, v_pipeline_id,
        jsonb_build_object('entered_at', now()));

    IF p_assigned_to_id IS NOT NULL THEN
        INSERT INTO edges (edge_type, source_id, target_id, properties, effective_from)
        VALUES ('assigned_to', v_opp_id, p_assigned_to_id, '{"role": "owner"}', now());
    END IF;

    RETURN v_opp_id;
END;
$$ LANGUAGE plpgsql;

5.2 Advance Deal Stage

CREATE FUNCTION advance_deal_stage(
    p_opp_id uuid,
    p_new_stage text,
    p_reason text DEFAULT NULL,
    p_actor text DEFAULT NULL
) RETURNS uuid AS $$
DECLARE
    v_old_assertion_id uuid;
    v_old_stage text;
    v_pipeline text;
    v_new_assertion_id uuid;
    v_event_id uuid;
BEGIN
    SELECT id, claim->>'stage', claim->>'pipeline'
    INTO v_old_assertion_id, v_old_stage, v_pipeline
    FROM current_assertions
    WHERE subject_node_id = p_opp_id AND assertion_type = 'deal_stage';

    INSERT INTO events (event_type, occurred_at, summary, properties, actor_system)
    VALUES (
        'stage_change', now(),
        format('Stage: %s -> %s', v_old_stage, p_new_stage),
        jsonb_build_object('from_stage', v_old_stage, 'to_stage', p_new_stage, 'reason', p_reason),
        coalesce(p_actor, current_setting('app.current_user_id', true))
    )
    RETURNING id INTO v_event_id;

    INSERT INTO event_participants (event_id, node_id, role)
    VALUES (v_event_id, p_opp_id, 'subject');

    v_new_assertion_id := supersede_assertion(
        p_old_assertion_id := v_old_assertion_id,
        p_new_assertion_type := 'deal_stage',
        p_new_subject_node_id := p_opp_id,
        p_new_subject_edge_id := NULL,
        p_new_claim := jsonb_build_object(
            'stage', p_new_stage, 'pipeline', v_pipeline,
            'moved_from', v_old_stage, 'reason', p_reason),
        p_new_source_event_id := v_event_id,
        p_new_confidence := 1.0
    );

    RETURN v_new_assertion_id;
END;
$$ LANGUAGE plpgsql;

5.3 Log Activity

CREATE FUNCTION log_crm_activity(
    p_event_type text,
    p_summary text,
    p_properties jsonb,
    p_actor text,
    p_participant_ids uuid[],
    p_participant_roles text[],
    p_occurred_at timestamptz DEFAULT now()
) RETURNS uuid AS $$
DECLARE
    v_event_id uuid;
    i int;
BEGIN
    INSERT INTO events (event_type, occurred_at, summary, properties, actor_system)
    VALUES (p_event_type, p_occurred_at, p_summary, p_properties, p_actor)
    RETURNING id INTO v_event_id;

    FOR i IN 1..array_length(p_participant_ids, 1) LOOP
        INSERT INTO event_participants (event_id, node_id, role)
        VALUES (v_event_id, p_participant_ids[i], p_participant_roles[i])
        ON CONFLICT (event_id, node_id, role) DO NOTHING;
    END LOOP;

    RETURN v_event_id;
END;
$$ LANGUAGE plpgsql;

6. Materialized Views

6.1 Active Opportunities

Flattens the graph into a traditional CRM table. Uses DISTINCT ON to prevent fan-out from multiple contacts or assignments per opportunity.

CREATE MATERIALIZED VIEW opportunities_active AS
SELECT
    n.id AS node_id,
    n.label,
    n.properties->>'code' AS code,
    n.properties->>'name' AS name,
    n.properties->>'estimated_value' AS estimated_value,
    n.attrs->'teams' AS teams,
    stg.claim->>'stage' AS stage,
    stg.claim->>'pipeline' AS pipeline,
    stg.asserted_at AS stage_since,
    val.claim->>'amount' AS current_value,
    wp.claim->>'probability' AS win_probability,
    pc.label AS primary_contact_name,
    pc.id AS primary_contact_id,
    owner.label AS assigned_to_name,
    owner.id AS assigned_to_id,
    n.created_at
FROM nodes n
LEFT JOIN current_assertions stg
    ON stg.subject_node_id = n.id AND stg.assertion_type = 'deal_stage'
LEFT JOIN current_assertions val
    ON val.subject_node_id = n.id AND val.assertion_type = 'deal_value'
LEFT JOIN current_assertions wp
    ON wp.subject_node_id = n.id AND wp.assertion_type = 'win_probability'
LEFT JOIN LATERAL (
    SELECT pc_n.id, pc_n.label
    FROM edges pc_e
    JOIN nodes pc_n ON pc_n.id = pc_e.target_id
    WHERE pc_e.source_id = n.id
      AND pc_e.edge_type = 'primary_contact'
      AND pc_e.archived_at IS NULL
    ORDER BY pc_e.created_at
    LIMIT 1
) pc ON true
LEFT JOIN LATERAL (
    SELECT own_n.id, own_n.label
    FROM edges own_e
    JOIN nodes own_n ON own_n.id = own_e.target_id
    WHERE own_e.source_id = n.id
      AND own_e.edge_type = 'assigned_to'
      AND own_e.properties->>'role' = 'owner'
      AND own_e.archived_at IS NULL
      AND own_e.effective_to IS NULL
    ORDER BY own_e.effective_from DESC
    LIMIT 1
) owner ON true
WHERE n.node_type = 'opportunity'
    AND n.archived_at IS NULL
    AND (stg.claim->>'stage' IS NULL
         OR stg.claim->>'stage' NOT IN ('closed_won', 'closed_lost', 'dead'));

CREATE UNIQUE INDEX idx_oa_node ON opportunities_active (node_id);
CREATE INDEX idx_oa_code ON opportunities_active (code);
CREATE INDEX idx_oa_stage ON opportunities_active (stage);
CREATE INDEX idx_oa_owner ON opportunities_active (assigned_to_id);

6.2 Contacts Directory

CREATE MATERIALIZED VIEW contacts_directory AS
SELECT
    n.id AS node_id,
    n.label,
    n.properties->>'code' AS code,
    n.properties->>'first_name' AS first_name,
    n.properties->>'last_name' AS last_name,
    n.properties->>'email' AS email,
    n.properties->>'phone' AS phone,
    n.properties->>'title' AS title,
    org.label AS organization,
    org.id AS org_id,
    ci.claim AS current_contact_info,
    sent.claim AS current_sentiment,
    n.created_at
FROM nodes n
LEFT JOIN LATERAL (
    SELECT o.id, o.label
    FROM edges emp ON emp.target_id = n.id
        AND emp.edge_type = 'employs'
        AND emp.archived_at IS NULL
        AND emp.effective_to IS NULL
    JOIN nodes o ON o.id = emp.source_id
    LIMIT 1
) org ON true
LEFT JOIN current_assertions ci
    ON ci.subject_node_id = n.id AND ci.assertion_type = 'contact_info'
LEFT JOIN current_assertions sent
    ON sent.subject_node_id = n.id AND sent.assertion_type = 'sentiment'
WHERE n.node_type = 'person'
    AND n.archived_at IS NULL;

CREATE UNIQUE INDEX idx_cd_node ON contacts_directory (node_id);
CREATE INDEX idx_cd_code ON contacts_directory (code);
CREATE INDEX idx_cd_name ON contacts_directory (last_name, first_name);
CREATE INDEX idx_cd_org ON contacts_directory (org_id);

6.3 Refresh

REFRESH MATERIALIZED VIEW CONCURRENTLY opportunities_active;
REFRESH MATERIALIZED VIEW CONCURRENTLY contacts_directory;

Guidance: Start with regular views. Promote to materialized views only after measuring query pressure.

7. Access Control

CRM access builds on the core RLS infrastructure (see Security).

7.1 Core Principles

Node visibility is the anchor. Can’t see the opportunity node? Can’t see its edges, events, or assertions.
Team-based by default. Opportunities are tagged with teams in attrs.
Role-based for sensitive data. Financial terms, negotiation stances require specific roles.

7.2 Example Access Grants

-- Sales team can read/write their opportunities
INSERT INTO access_grants (grantee, grant_type, resource_type, access_level, scope) VALUES
    ('sales', 'team', 'node', 'write', '{"node_type": "opportunity", "teams": ["sales"]}');

-- Support team can read all customers (cross-team for ticket context)
INSERT INTO access_grants (grantee, grant_type, resource_type, access_level, scope) VALUES
    ('support', 'team', 'node', 'read', '{"node_type": "person"}');

-- Finance can see deal values across all opportunities
INSERT INTO access_grants (grantee, grant_type, resource_type, access_level, scope) VALUES
    ('finance', 'team', 'assertion', 'read', '{"assertion_type": "deal_value"}');

8. Cross-Domain Integration

CRM entities exist in the same graph as everything else. An opportunity targeting a customer account is an edge. A contact who filed a support ticket is the same person node.

-- "Show me all open support tickets for contacts on active opportunities"
SELECT
    opp.properties->>'code' AS opp_code,
    con.label AS contact_name,
    ticket.label AS ticket_title,
    ticket_stg.claim->>'status' AS ticket_status
FROM nodes opp
JOIN edges pc ON pc.source_id = opp.id
    AND pc.edge_type IN ('primary_contact', 'secondary_contact')
    AND pc.archived_at IS NULL
JOIN nodes con ON con.id = pc.target_id
JOIN edges filed ON filed.source_id = con.id
    AND filed.edge_type = 'filed'
    AND filed.archived_at IS NULL
JOIN nodes ticket ON ticket.id = filed.target_id AND ticket.node_type = 'ticket'
LEFT JOIN current_assertions ticket_stg
    ON ticket_stg.subject_node_id = ticket.id AND ticket_stg.assertion_type = 'ticket_status'
WHERE opp.node_type = 'opportunity' AND opp.archived_at IS NULL
    AND (ticket_stg.claim->>'status' NOT IN ('resolved', 'closed')
         OR ticket_stg.claim->>'status' IS NULL);