Graph Evolution & Temporal Knowledge

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

-- Temporal entity structure
CREATE TABLE temporal_entities (
    entity_id UUID PRIMARY KEY,
    version_id UUID NOT NULL,
    parent_version_id UUID REFERENCES temporal_entities(version_id),
    valid_from TIMESTAMPTZ DEFAULT NOW(),
    valid_to TIMESTAMPTZ DEFAULT 'infinity',
    created_by UUID REFERENCES auth.users(id),
    entity_type TEXT NOT NULL,
    content JSONB NOT NULL,
    embeddings vector(1536),
    metadata JSONB DEFAULT '{}'::jsonb,
    is_deleted BOOLEAN DEFAULT FALSE,
    
    -- Temporal constraints
    CONSTRAINT valid_temporal_range CHECK (valid_from < valid_to),
    CONSTRAINT no_overlap_versions EXCLUDE USING gist (
        entity_id WITH =,
        tstzrange(valid_from, valid_to) WITH &&
    ) WHERE (NOT is_deleted)
);

-- Temporal relationship tracking
CREATE TABLE temporal_relationships (
    relationship_id UUID PRIMARY KEY,
    source_entity_id UUID NOT NULL,
    target_entity_id UUID NOT NULL,
    relationship_type TEXT NOT NULL,
    valid_from TIMESTAMPTZ DEFAULT NOW(),
    valid_to TIMESTAMPTZ DEFAULT 'infinity',
    strength FLOAT DEFAULT 1.0,
    confidence FLOAT DEFAULT 1.0,
    metadata JSONB DEFAULT '{}'::jsonb,
    
    -- Foreign keys to temporal entities
    FOREIGN KEY (source_entity_id) REFERENCES temporal_entities(entity_id),
    FOREIGN KEY (target_entity_id) REFERENCES temporal_entities(entity_id)
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

-- Recursive CTE for temporal graph traversal
CREATE OR REPLACE FUNCTION traverse_temporal_graph(
    start_entity_id UUID,
    target_timestamp TIMESTAMPTZ DEFAULT NOW(),
    max_depth INTEGER DEFAULT 5,
    relationship_types TEXT[] DEFAULT NULL
) RETURNS TABLE (
    entity_id UUID,
    path UUID[],
    depth INTEGER,
    total_strength FLOAT
) AS $$
WITH RECURSIVE temporal_traverse AS (
    -- Base case: starting entity
    SELECT 
        te.entity_id,
        ARRAY[te.entity_id] as path,
        0 as depth,
        1.0 as total_strength
    FROM temporal_entities te
    WHERE te.entity_id = start_entity_id
        AND te.valid_from <= target_timestamp
        AND te.valid_to > target_timestamp
        AND NOT te.is_deleted
    
    UNION ALL
    
    -- Recursive case: follow relationships
    SELECT 
        tr.target_entity_id,
        tt.path || tr.target_entity_id,
        tt.depth + 1,
        tt.total_strength * tr.strength
    FROM temporal_traverse tt
    JOIN temporal_relationships tr ON tt.entity_id = tr.source_entity_id
    JOIN temporal_entities te ON tr.target_entity_id = te.entity_id
    WHERE tt.depth < max_depth
        AND tr.valid_from <= target_timestamp
        AND tr.valid_to > target_timestamp
        AND te.valid_from <= target_timestamp
        AND te.valid_to > target_timestamp
        AND NOT te.is_deleted
        AND tr.target_entity_id != ALL(tt.path) -- Prevent cycles
        AND (relationship_types IS NULL OR tr.relationship_type = ANY(relationship_types))
)
SELECT * FROM temporal_traverse;
$$ LANGUAGE SQL;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

-- Time-travel query function
CREATE OR REPLACE FUNCTION knowledge_at_time(
    query_timestamp TIMESTAMPTZ,
    entity_filter JSONB DEFAULT '{}'::jsonb
) RETURNS TABLE (
    entity_id UUID,
    content JSONB,
    relationships JSONB
) AS $$
BEGIN
    RETURN QUERY
    WITH entities_at_time AS (
        SELECT 
            te.entity_id,
            te.content,
            te.metadata
        FROM temporal_entities te
        WHERE te.valid_from <= query_timestamp
            AND te.valid_to > query_timestamp
            AND NOT te.is_deleted
            AND (entity_filter = '{}'::jsonb OR te.content @> entity_filter)
    ),
    relationships_at_time AS (
        SELECT 
            tr.source_entity_id,
            jsonb_agg(
                jsonb_build_object(
                    'target', tr.target_entity_id,
                    'type', tr.relationship_type,
                    'strength', tr.strength,
                    'confidence', tr.confidence
                )
            ) as relationships
        FROM temporal_relationships tr
        WHERE tr.valid_from <= query_timestamp
            AND tr.valid_to > query_timestamp
        GROUP BY tr.source_entity_id
    )
    SELECT 
        eat.entity_id,
        eat.content,
        COALESCE(rat.relationships, '[]'::jsonb) as relationships
    FROM entities_at_time eat
    LEFT JOIN relationships_at_time rat ON eat.entity_id = rat.source_entity_id;
END;
$$ LANGUAGE plpgsql;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

-- Knowledge ledger for complete audit trail
CREATE TABLE knowledge_ledger (
    ledger_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    block_number BIGSERIAL,
    previous_hash TEXT,
    current_hash TEXT NOT NULL,
    timestamp TIMESTAMPTZ DEFAULT NOW(),
    operation_type TEXT NOT NULL, -- 'CREATE', 'UPDATE', 'DELETE', 'MERGE'
    entity_id UUID NOT NULL,
    user_id UUID REFERENCES auth.users(id),
    changes JSONB NOT NULL,
    signature TEXT, -- Cryptographic signature
    
    -- Blockchain-style integrity
    CONSTRAINT valid_operation CHECK (operation_type IN ('CREATE', 'UPDATE', 'DELETE', 'MERGE', 'FORK'))
);

-- Function to compute hash of ledger entry
CREATE OR REPLACE FUNCTION compute_ledger_hash(
    block_number BIGINT,
    previous_hash TEXT,
    operation_type TEXT,
    entity_id UUID,
    changes JSONB,
    timestamp TIMESTAMPTZ
) RETURNS TEXT AS $$
BEGIN
    RETURN encode(
        digest(
            block_number::text || 
            COALESCE(previous_hash, '') || 
            operation_type || 
            entity_id::text || 
            changes::text || 
            extract(epoch from timestamp)::text,
            'sha256'
        ),
        'hex'
    );
END;
$$ LANGUAGE plpgsql;

-- Trigger to maintain ledger integrity
CREATE OR REPLACE FUNCTION maintain_ledger_integrity()
RETURNS TRIGGER AS $$
DECLARE
    prev_hash TEXT;
    computed_hash TEXT;
BEGIN
    -- Get previous hash
    SELECT current_hash INTO prev_hash
    FROM knowledge_ledger
    ORDER BY block_number DESC
    LIMIT 1;
    
    -- Compute current hash
    computed_hash := compute_ledger_hash(
        NEW.block_number,
        prev_hash,
        NEW.operation_type,
        NEW.entity_id,
        NEW.changes,
        NEW.timestamp
    );
    
    NEW.previous_hash := prev_hash;
    NEW.current_hash := computed_hash;
    
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER ledger_integrity_trigger
    BEFORE INSERT ON knowledge_ledger
    FOR EACH ROW
    EXECUTE FUNCTION maintain_ledger_integrity();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

interface KnowledgeLineage {
  entityId: string;
  genealogy: {
    ancestors: LineageNode[];
    descendants: LineageNode[];
    siblings: LineageNode[]; // Same parent versions
  };
  evolutionPath: {
    created: Date;
    modifications: ModificationEvent[];
    branches: BranchEvent[];
    merges: MergeEvent[];
  };
}

interface LineageNode {
  entityId: string;
  versionId: string;
  timestamp: Date;
  operation: 'CREATE' | 'UPDATE' | 'FORK' | 'MERGE' | 'DELETE';
  author: string;
  changes: ChangeSet;
  confidence: number;
}

class TemporalKnowledgeManager {
  async traceLineage(entityId: string): Promise<KnowledgeLineage> {
    const { data, error } = await supabase.rpc('trace_entity_lineage', {
      entity_id: entityId
    });

    if (error) throw error;

    return {
      entityId,
      genealogy: await this.buildGenealogyTree(data.ancestors, data.descendants),
      evolutionPath: await this.buildEvolutionPath(data.history)
    };
  }

  private async buildGenealogyTree(ancestors: any[], descendants: any[]) {
    // Build complete family tree of knowledge entity
    const ancestorNodes = ancestors.map(a => ({
      entityId: a.entity_id,
      versionId: a.version_id,
      timestamp: new Date(a.created_at),
      operation: a.operation_type,
      author: a.created_by,
      changes: JSON.parse(a.changes),
      confidence: a.confidence_score
    }));

    const descendantNodes = descendants.map(d => ({
      entityId: d.entity_id,
      versionId: d.version_id,
      timestamp: new Date(d.created_at),
      operation: d.operation_type,
      author: d.created_by,
      changes: JSON.parse(d.changes),
      confidence: d.confidence_score
    }));

    // Find sibling versions (same parent)
    const siblings = await this.findSiblingVersions(entityId);

    return { ancestors: ancestorNodes, descendants: descendantNodes, siblings };
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

-- Knowledge branches for experimentation
CREATE TABLE knowledge_branches (
    branch_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    branch_name TEXT NOT NULL,
    parent_branch_id UUID REFERENCES knowledge_branches(branch_id),
    created_by UUID REFERENCES auth.users(id),
    created_at TIMESTAMPTZ DEFAULT NOW(),
    description TEXT,
    is_experimental BOOLEAN DEFAULT TRUE,
    merge_status TEXT DEFAULT 'ACTIVE', -- 'ACTIVE', 'MERGED', 'ABANDONED'
    
    UNIQUE(branch_name, created_by)
);

-- Branch-specific entity versions
CREATE TABLE branch_entities (
    branch_entity_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    branch_id UUID REFERENCES knowledge_branches(branch_id),
    entity_id UUID REFERENCES temporal_entities(entity_id),
    branch_version JSONB NOT NULL,
    conflict_resolution JSONB, -- How conflicts were resolved
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    UNIQUE(branch_id, entity_id)
);

-- Fork operation
CREATE OR REPLACE FUNCTION fork_knowledge_branch(
    source_branch_id UUID,
    new_branch_name TEXT,
    user_id UUID,
    description TEXT DEFAULT NULL
) RETURNS UUID AS $$
DECLARE
    new_branch_id UUID;
BEGIN
    -- Create new branch
    INSERT INTO knowledge_branches (branch_name, parent_branch_id, created_by, description)
    VALUES (new_branch_name, source_branch_id, user_id, description)
    RETURNING branch_id INTO new_branch_id;
    
    -- Copy all entities from source branch
    INSERT INTO branch_entities (branch_id, entity_id, branch_version)
    SELECT 
        new_branch_id,
        be.entity_id,
        be.branch_version
    FROM branch_entities be
    WHERE be.branch_id = source_branch_id;
    
    -- Log the fork operation
    INSERT INTO knowledge_ledger (operation_type, entity_id, user_id, changes)
    VALUES ('FORK', new_branch_id, user_id, jsonb_build_object(
        'source_branch', source_branch_id,
        'new_branch', new_branch_id,
        'branch_name', new_branch_name
    ));
    
    RETURN new_branch_id;
END;
$$ LANGUAGE plpgsql;

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

interface MergeStrategy {
  name: string;
  description: string;
  conflictResolver: (conflicts: Conflict[]) => Resolution[];
}

class KnowledgeMerger {
  private strategies: Map<string, MergeStrategy> = new Map([
    ['semantic-similarity', {
      name: 'Semantic Similarity Merge',
      description: 'Use vector embeddings to resolve semantic conflicts',
      conflictResolver: this.semanticMergeResolver.bind(this)
    }],
    ['community-consensus', {
      name: 'Community Consensus',
      description: 'Use voting and reputation to resolve conflicts',
      conflictResolver: this.consensusMergeResolver.bind(this)
    }],
    ['temporal-priority', {
      name: 'Temporal Priority',
      description: 'Prefer newer knowledge with decay factor',
      conflictResolver: this.temporalMergeResolver.bind(this)
    }],
    ['ml-automated', {
      name: 'ML Automated Resolution',
      description: 'Use trained models to predict best resolution',
      conflictResolver: this.mlMergeResolver.bind(this)
    }]
  ]);

  async mergeBranches(
    sourceBranchId: string,
    targetBranchId: string,
    strategy: string = 'semantic-similarity'
  ): Promise<MergeResult> {
    const conflicts = await this.detectConflicts(sourceBranchId, targetBranchId);
    
    if (conflicts.length === 0) {
      return this.fastForwardMerge(sourceBranchId, targetBranchId);
    }

    const mergeStrategy = this.strategies.get(strategy);
    if (!mergeStrategy) {
      throw new Error(`Unknown merge strategy: ${strategy}`);
    }

    const resolutions = await mergeStrategy.conflictResolver(conflicts);
    return this.applyMergeResolutions(sourceBranchId, targetBranchId, resolutions);
  }

  private async semanticMergeResolver(conflicts: Conflict[]): Promise<Resolution[]> {
    const resolutions: Resolution[] = [];
    
    for (const conflict of conflicts) {
      // Use vector similarity to determine best merge
      const sourceEmbedding = await this.getEntityEmbedding(conflict.sourceVersion);
      const targetEmbedding = await this.getEntityEmbedding(conflict.targetVersion);
      const contextEmbedding = await this.getContextEmbedding(conflict.context);
      
      const sourceSimilarity = this.cosineSimilarity(sourceEmbedding, contextEmbedding);
      const targetSimilarity = this.cosineSimilarity(targetEmbedding, contextEmbedding);
      
      if (sourceSimilarity > targetSimilarity + 0.1) {
        resolutions.push({
          conflictId: conflict.id,
          resolution: 'USE_SOURCE',
          confidence: sourceSimilarity,
          reasoning: 'Higher semantic similarity to context'
        });
      } else if (targetSimilarity > sourceSimilarity + 0.1) {
        resolutions.push({
          conflictId: conflict.id,
          resolution: 'USE_TARGET',
          confidence: targetSimilarity,
          reasoning: 'Higher semantic similarity to context'
        });
      } else {
        // Create hybrid version
        const hybridContent = await this.createHybridContent(
          conflict.sourceVersion,
          conflict.targetVersion,
          contextEmbedding
        );
        
        resolutions.push({
          conflictId: conflict.id,
          resolution: 'CREATE_HYBRID',
          content: hybridContent,
          confidence: Math.max(sourceSimilarity, targetSimilarity),
          reasoning: 'Semantic hybrid of both versions'
        });
      }
    }
    
    return resolutions;
  }

  private async consensusMergeResolver(conflicts: Conflict[]): Promise<Resolution[]> {
    const resolutions: Resolution[] = [];
    
    for (const conflict of conflicts) {
      // Get community votes and expert opinions
      const votes = await supabase
        .from('conflict_votes')
        .select('*')
        .eq('conflict_id', conflict.id);
      
      const expertOpinions = await supabase
        .from('expert_opinions')
        .select('*')
        .eq('conflict_id', conflict.id);
      
      // Weight votes by user reputation
      let sourceScore = 0;
      let targetScore = 0;
      let totalWeight = 0;
      
      for (const vote of votes.data || []) {
        const userReputation = await this.getUserReputation(vote.user_id);
        const weight = Math.log(userReputation + 1);
        
        if (vote.preference === 'SOURCE') {
          sourceScore += weight;
        } else if (vote.preference === 'TARGET') {
          targetScore += weight;
        }
        
        totalWeight += weight;
      }
      
      // Expert opinions carry higher weight
      for (const opinion of expertOpinions.data || []) {
        const expertWeight = 10; // High weight for experts
        
        if (opinion.recommendation === 'SOURCE') {
          sourceScore += expertWeight;
        } else if (opinion.recommendation === 'TARGET') {
          targetScore += expertWeight;
        }
        
        totalWeight += expertWeight;
      }
      
      if (totalWeight === 0) {
        // No community input, fall back to temporal priority
        resolutions.push(...await this.temporalMergeResolver([conflict]));
      } else {
        const confidence = Math.abs(sourceScore - targetScore) / totalWeight;
        
        resolutions.push({
          conflictId: conflict.id,
          resolution: sourceScore > targetScore ? 'USE_SOURCE' : 'USE_TARGET',
          confidence,
          reasoning: `Community consensus (${Math.round(confidence * 100)}% agreement)`
        });
      }
    }
    
    return resolutions;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

-- Knowledge decay function
CREATE OR REPLACE FUNCTION calculate_knowledge_decay(
    entity_id UUID,
    base_timestamp TIMESTAMPTZ DEFAULT NOW()
) RETURNS FLOAT AS $$
DECLARE
    entity_age INTERVAL;
    last_update TIMESTAMPTZ;
    update_frequency FLOAT;
    domain_half_life INTERVAL;
    base_confidence FLOAT;
    decay_factor FLOAT;
BEGIN
    -- Get entity metadata
    SELECT 
        valid_from,
        (metadata->>'confidence')::float,
        (metadata->>'update_frequency')::float,
        (metadata->>'domain')::text
    INTO last_update, base_confidence, update_frequency, domain_type
    FROM temporal_entities
    WHERE entity_id = entity_id
    AND valid_to = 'infinity'
    AND NOT is_deleted;
    
    -- Calculate age
    entity_age := base_timestamp - last_update;
    
    -- Domain-specific half-life
    domain_half_life := CASE 
        WHEN domain_type = 'technology' THEN INTERVAL '6 months'
        WHEN domain_type = 'science' THEN INTERVAL '2 years'
        WHEN domain_type = 'finance' THEN INTERVAL '3 months'
        WHEN domain_type = 'general' THEN INTERVAL '1 year'
        ELSE INTERVAL '1 year'
    END;
    
    -- Calculate decay using exponential decay formula
    decay_factor := base_confidence * EXP(-LN(2) * EXTRACT(EPOCH FROM entity_age) / EXTRACT(EPOCH FROM domain_half_life));
    
    -- Factor in update frequency (frequently updated knowledge decays slower)
    IF update_frequency > 0 THEN
        decay_factor := decay_factor * (1 + LOG(update_frequency + 1) * 0.1);
    END IF;
    
    -- Ensure bounds [0, 1]
    RETURN GREATEST(0, LEAST(1, decay_factor));
END;
$$ LANGUAGE plpgsql;

-- Automated decay monitoring
CREATE OR REPLACE FUNCTION monitor_knowledge_decay()
RETURNS VOID AS $$
BEGIN
    -- Update confidence scores based on decay
    UPDATE temporal_entities
    SET metadata = metadata || jsonb_build_object(
        'current_confidence', calculate_knowledge_decay(entity_id),
        'last_decay_check', NOW()
    )
    WHERE valid_to = 'infinity'
    AND NOT is_deleted
    AND (metadata->>'last_decay_check')::timestamptz < NOW() - INTERVAL '1 day';
    
    -- Flag entities with very low confidence for review
    INSERT INTO knowledge_review_queue (entity_id, reason, priority)
    SELECT 
        entity_id,
        'Knowledge decay below threshold',
        CASE 
            WHEN (metadata->>'current_confidence')::float < 0.1 THEN 'HIGH'
            WHEN (metadata->>'current_confidence')::float < 0.3 THEN 'MEDIUM'
            ELSE 'LOW'
        END
    FROM temporal_entities
    WHERE valid_to = 'infinity'
    AND NOT is_deleted
    AND (metadata->>'current_confidence')::float < 0.5
    AND entity_id NOT IN (SELECT entity_id FROM knowledge_review_queue WHERE status = 'PENDING');
END;
$$ LANGUAGE plpgsql;

-- Schedule decay monitoring
SELECT cron.schedule('knowledge-decay-monitor', '0 2 * * *', 'SELECT monitor_knowledge_decay();');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

interface KnowledgeEvent {
  eventId: string;
  eventType: string;
  entityId: string;
  userId: string;
  timestamp: Date;
  data: any;
  metadata: {
    causationId?: string; // What caused this event
    correlationId?: string; // Group related events
    version: number;
    checksum: string;
  };
}

class EventSourcingManager {
  async appendEvent(event: Omit<KnowledgeEvent, 'eventId' | 'timestamp'>): Promise<string> {
    const eventId = crypto.randomUUID();
    const timestamp = new Date();
    const checksum = this.calculateChecksum(event.data);
    
    const fullEvent: KnowledgeEvent = {
      ...event,
      eventId,
      timestamp,
      metadata: {
        ...event.metadata,
        checksum
      }
    };
    
    // Store in event store
    const { error } = await supabase
      .from('knowledge_events')
      .insert({
        event_id: eventId,
        event_type: event.eventType,
        entity_id: event.entityId,
        user_id: event.userId,
        timestamp,
        data: event.data,
        metadata: fullEvent.metadata
      });
    
    if (error) throw error;
    
    // Update read model
    await this.updateReadModel(fullEvent);
    
    // Trigger event handlers
    await this.triggerEventHandlers(fullEvent);
    
    return eventId;
  }
  
  async replayEvents(entityId: string, fromVersion?: number): Promise<any> {
    const { data: events } = await supabase
      .from('knowledge_events')
      .select('*')
      .eq('entity_id', entityId)
      .gte('metadata->>version', fromVersion || 0)
      .order('timestamp');
    
    let state = {};
    
    for (const event of events || []) {
      state = this.applyEvent(state, event);
    }
    
    return state;
  }
  
  private applyEvent(state: any, event: any): any {
    switch (event.event_type) {
      case 'ENTITY_CREATED':
        return { ...event.data };
        
      case 'ENTITY_UPDATED':
        return { ...state, ...event.data.changes };
        
      case 'RELATIONSHIP_ADDED':
        return {
          ...state,
          relationships: [...(state.relationships || []), event.data.relationship]
        };
        
      case 'ENTITY_MERGED':
        return { ...state, ...event.data.mergedContent };
        
      default:
        return state;
    }
  }
}