Compare commits
2 Commits
5afd3232cb
...
65a6230959
Author | SHA1 | Date | |
---|---|---|---|
65a6230959 | |||
07c8da3cea |
@ -436,95 +436,6 @@ describe('Custom Resolvers', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('Plugin Communication', () => {
|
||||
test('plugins should be able to access each other\'s states', () => {
|
||||
// Create a plugin that depends on another property's value
|
||||
class DependentPlugin implements ResolverPlugin<{ value?: string }> {
|
||||
name = 'dependent';
|
||||
|
||||
initialize() {
|
||||
return { value: 'initial' };
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { value?: string },
|
||||
_newValue: PropertyTypes,
|
||||
_delta: CollapsedDelta,
|
||||
context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
// This plugin's value depends on the 'source' property's resolved value
|
||||
const sourceValue = context?.resolvedValues['source'];
|
||||
return {
|
||||
value: typeof sourceValue === 'string' ? `Processed: ${sourceValue}` : currentState.value
|
||||
};
|
||||
}
|
||||
|
||||
resolve(
|
||||
state: { value?: string },
|
||||
context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
): PropertyTypes | undefined {
|
||||
// In resolve, we can also check the context if needed
|
||||
const sourceValue = context?.resolvedValues['source'];
|
||||
if (typeof sourceValue === 'string' && state.value === 'initial') {
|
||||
return `Processed: ${sourceValue}`;
|
||||
}
|
||||
return state.value;
|
||||
}
|
||||
}
|
||||
|
||||
// Create a resolver with both plugins
|
||||
const resolver = new CustomResolver(lossless, {
|
||||
source: new LastWriteWinsPlugin(),
|
||||
dependent: new DependentPlugin()
|
||||
});
|
||||
|
||||
// First, set up the source property
|
||||
const sourceDelta = createDelta('user1', 'host1')
|
||||
.withTimestamp(1000)
|
||||
.setProperty('entity1', 'source', 'original', 'collection')
|
||||
.buildV1();
|
||||
|
||||
lossless.ingestDelta(sourceDelta);
|
||||
|
||||
// Then set up the dependent property
|
||||
const dependentDelta = createDelta('user1', 'host1')
|
||||
.withTimestamp(2000)
|
||||
.setProperty('entity1', 'dependent', 'initial', 'collection')
|
||||
.buildV1();
|
||||
|
||||
lossless.ingestDelta(dependentDelta);
|
||||
|
||||
// Get the first result
|
||||
const result = resolver.resolve();
|
||||
expect(result).toBeDefined();
|
||||
|
||||
// The dependent plugin should see the source value
|
||||
expect(result!['entity1'].properties).toMatchObject({
|
||||
source: 'original',
|
||||
dependent: expect.stringContaining('Processed: original')
|
||||
});
|
||||
|
||||
// Create a new delta that updates the source property
|
||||
const updateDelta = createDelta('user1', 'host1')
|
||||
.withTimestamp(3000)
|
||||
.setProperty('entity1', 'source', 'updated', 'collection')
|
||||
.buildV1();
|
||||
|
||||
// Ingest the update delta
|
||||
lossless.ingestDelta(updateDelta);
|
||||
|
||||
// Get the updated result
|
||||
const updatedResult = resolver.resolve();
|
||||
expect(updatedResult).toBeDefined();
|
||||
|
||||
// The dependent plugin should see the updated source value
|
||||
expect(updatedResult!['entity1'].properties).toMatchObject({
|
||||
source: 'updated',
|
||||
dependent: expect.stringContaining('Processed: updated')
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('Edge Cases', () => {
|
||||
test('should handle empty delta sets', () => {
|
||||
const resolver = new CustomResolver(lossless, {
|
||||
|
@ -1,67 +0,0 @@
|
||||
import { RhizomeNode, Lossless } from "../../../../src";
|
||||
import { RelationshipGraphResolver } from "../../../../src/views/resolvers/relationship-graph";
|
||||
import { createDelta } from "../../../../src/core/delta-builder";
|
||||
|
||||
// Deferring until we figure out lossy view composition
|
||||
describe.skip('RelationshipGraphResolver', () => {
|
||||
let node: RhizomeNode;
|
||||
let lossless: Lossless;
|
||||
let resolver: RelationshipGraphResolver;
|
||||
|
||||
beforeEach(() => {
|
||||
node = new RhizomeNode();
|
||||
lossless = new Lossless(node);
|
||||
resolver = new RelationshipGraphResolver(lossless);
|
||||
});
|
||||
|
||||
describe('Basic Relationship Resolution', () => {
|
||||
test('should create a basic relationship between two entities', () => {
|
||||
const delta = createDelta('user1', 'host1')
|
||||
.withId('delta1')
|
||||
.withTimestamp(1000)
|
||||
.relate('entity1', 'entity2', 'friends', { since: 1640995200000 })
|
||||
.buildV1();
|
||||
|
||||
// Ingest the delta
|
||||
lossless.ingestDelta(delta);
|
||||
|
||||
// Resolve the graph
|
||||
const graph = resolver.resolve();
|
||||
|
||||
// Verify the relationship was created
|
||||
expect(graph).toBeDefined();
|
||||
if (!graph) return;
|
||||
|
||||
console.log(JSON.stringify(graph, null, 2));
|
||||
|
||||
// Check entities exist
|
||||
expect(graph.entities.has('entity1')).toBe(true);
|
||||
expect(graph.entities.has('entity2')).toBe(true);
|
||||
|
||||
// Check relationship exists
|
||||
const relationshipId = 'delta1';
|
||||
expect(graph.relationships.has(relationshipId)).toBe(true);
|
||||
|
||||
const relationship = graph.relationships.get(relationshipId)!;
|
||||
expect(relationship).toEqual({
|
||||
id: relationshipId,
|
||||
type: 'friends',
|
||||
source: 'entity1',
|
||||
target: 'entity2',
|
||||
properties: { since: 1640995200000 },
|
||||
deltas: ['delta1']
|
||||
});
|
||||
|
||||
// Check entity relationships
|
||||
const entity1 = graph.entities.get('entity1')!;
|
||||
const entity2 = graph.entities.get('entity2')!;
|
||||
|
||||
expect(entity1.outbound.has(relationshipId)).toBe(true);
|
||||
expect(entity2.inbound.has(relationshipId)).toBe(true);
|
||||
|
||||
// Check relationship type index
|
||||
expect(graph.relationshipsByType.has('friends')).toBe(true);
|
||||
expect(graph.relationshipsByType.get('friends')?.has(relationshipId)).toBe(true);
|
||||
});
|
||||
});
|
||||
});
|
@ -11,30 +11,11 @@ export interface ResolverPlugin<T = unknown> {
|
||||
initialize(): T;
|
||||
|
||||
// Process a new value for the property
|
||||
update(
|
||||
currentState: T,
|
||||
newValue: PropertyTypes,
|
||||
delta: CollapsedDelta,
|
||||
// Additional context including other properties' states
|
||||
context?: {
|
||||
// Current state of all properties for the entity
|
||||
entityState: Record<string, unknown>;
|
||||
// Current resolved values of all properties for the entity
|
||||
resolvedValues: Record<string, PropertyTypes>;
|
||||
}
|
||||
): T;
|
||||
update(currentState: T, newValue: PropertyTypes, delta: CollapsedDelta): T;
|
||||
|
||||
// Resolve the final value from the accumulated state
|
||||
resolve(
|
||||
state: T,
|
||||
// Additional context including other properties' states
|
||||
context?: {
|
||||
// Current state of all properties for the entity
|
||||
entityState: Record<string, unknown>;
|
||||
// Current resolved values of all properties for the entity
|
||||
resolvedValues: Record<string, PropertyTypes>;
|
||||
}
|
||||
): PropertyTypes | undefined;
|
||||
// Returns undefined if no valid value could be resolved
|
||||
resolve(state: T): PropertyTypes | undefined;
|
||||
}
|
||||
|
||||
// Configuration for custom resolver
|
||||
@ -94,15 +75,11 @@ export class CustomResolver extends Lossy<CustomResolverAccumulator, CustomResol
|
||||
acc[cur.id] = { id: cur.id, properties: {} };
|
||||
}
|
||||
|
||||
// First pass: collect all property states and resolved values
|
||||
const entityState: Record<string, unknown> = {};
|
||||
const resolvedValues: Record<string, PropertyTypes> = {};
|
||||
|
||||
// Initialize all properties first
|
||||
for (const propertyId of Object.keys(cur.propertyDeltas)) {
|
||||
for (const [propertyId, deltas] of Object.entries(cur.propertyDeltas)) {
|
||||
const plugin = this.config[propertyId];
|
||||
if (!plugin) continue;
|
||||
|
||||
// Initialize property state if not exists
|
||||
if (!acc[cur.id].properties[propertyId]) {
|
||||
acc[cur.id].properties[propertyId] = {
|
||||
plugin,
|
||||
@ -110,51 +87,13 @@ export class CustomResolver extends Lossy<CustomResolverAccumulator, CustomResol
|
||||
};
|
||||
}
|
||||
|
||||
// Store the current state
|
||||
entityState[propertyId] = acc[cur.id].properties[propertyId].state;
|
||||
|
||||
// Resolve current value if possible
|
||||
try {
|
||||
const resolved = plugin.resolve(acc[cur.id].properties[propertyId].state, {
|
||||
entityState: {},
|
||||
resolvedValues: {}
|
||||
});
|
||||
if (resolved !== undefined) {
|
||||
resolvedValues[propertyId] = resolved;
|
||||
}
|
||||
} catch (_e) {
|
||||
// Ignore resolution errors during reduction
|
||||
}
|
||||
}
|
||||
|
||||
// Second pass: process deltas with full context
|
||||
for (const [propertyId, deltas] of Object.entries(cur.propertyDeltas)) {
|
||||
const plugin = this.config[propertyId];
|
||||
if (!plugin) continue;
|
||||
|
||||
const propertyState = acc[cur.id].properties[propertyId];
|
||||
const context = { entityState, resolvedValues };
|
||||
|
||||
// Process all deltas for this property
|
||||
for (const delta of deltas || []) {
|
||||
const value = extractValueFromDelta(propertyId, delta);
|
||||
if (value !== undefined) {
|
||||
propertyState.state = plugin.update(
|
||||
propertyState.state,
|
||||
value,
|
||||
delta,
|
||||
context
|
||||
);
|
||||
|
||||
// Update the resolved value after each update
|
||||
try {
|
||||
const resolved = plugin.resolve(propertyState.state, context);
|
||||
if (resolved !== undefined) {
|
||||
resolvedValues[propertyId] = resolved;
|
||||
}
|
||||
} catch (_e) {
|
||||
// Ignore resolution errors during reduction
|
||||
}
|
||||
propertyState.state = propertyState.plugin.update(propertyState.state, value, delta);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -168,39 +107,11 @@ export class CustomResolver extends Lossy<CustomResolverAccumulator, CustomResol
|
||||
for (const [entityId, entity] of Object.entries(cur)) {
|
||||
const entityResult: { id: string; properties: EntityProperties } = { id: entityId, properties: {} };
|
||||
|
||||
// First pass: collect all property states
|
||||
const entityState: Record<string, unknown> = {};
|
||||
const resolvedValues: Record<string, PropertyTypes> = {};
|
||||
|
||||
// Initialize with current states and resolve all properties
|
||||
for (const [propertyId, propertyState] of Object.entries(entity.properties)) {
|
||||
entityState[propertyId] = propertyState.state;
|
||||
// Initial resolution with empty context
|
||||
try {
|
||||
const resolved = propertyState.plugin.resolve(propertyState.state, {
|
||||
entityState: {},
|
||||
resolvedValues: {}
|
||||
});
|
||||
if (resolved !== undefined) {
|
||||
resolvedValues[propertyId] = resolved;
|
||||
}
|
||||
} catch (_e) {
|
||||
// Ignore resolution errors
|
||||
}
|
||||
}
|
||||
|
||||
// Second pass: resolve with full context
|
||||
for (const [propertyId, propertyState] of Object.entries(entity.properties)) {
|
||||
const context = { entityState, resolvedValues };
|
||||
try {
|
||||
const resolvedValue = propertyState.plugin.resolve(propertyState.state, context);
|
||||
if (resolvedValue !== undefined) {
|
||||
entityResult.properties[propertyId] = resolvedValue;
|
||||
// Update the resolved value for dependent properties
|
||||
resolvedValues[propertyId] = resolvedValue;
|
||||
}
|
||||
} catch (_e) {
|
||||
// Ignore resolution errors
|
||||
const resolvedValue = propertyState.plugin.resolve(propertyState.state);
|
||||
// Only add the property if the resolved value is not undefined
|
||||
if (resolvedValue !== undefined) {
|
||||
entityResult.properties[propertyId] = resolvedValue;
|
||||
}
|
||||
}
|
||||
|
||||
@ -226,12 +137,7 @@ export class LastWriteWinsPlugin implements ResolverPlugin<{ value?: PropertyTyp
|
||||
return { timestamp: 0 };
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { value?: PropertyTypes, timestamp: number },
|
||||
newValue: PropertyTypes,
|
||||
delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
update(currentState: { value?: PropertyTypes, timestamp: number }, newValue: PropertyTypes, delta: CollapsedDelta) {
|
||||
if (delta.timeCreated > currentState.timestamp) {
|
||||
return {
|
||||
value: newValue,
|
||||
@ -254,12 +160,7 @@ export class FirstWriteWinsPlugin implements ResolverPlugin<{ value?: PropertyTy
|
||||
return { timestamp: Infinity };
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { value?: PropertyTypes, timestamp: number },
|
||||
newValue: PropertyTypes,
|
||||
delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
update(currentState: { value?: PropertyTypes, timestamp: number }, newValue: PropertyTypes, delta: CollapsedDelta) {
|
||||
if (delta.timeCreated < currentState.timestamp) {
|
||||
return {
|
||||
value: newValue,
|
||||
@ -284,12 +185,7 @@ export class ConcatenationPlugin implements ResolverPlugin<{ values: { value: st
|
||||
return { values: [] };
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { values: { value: string, timestamp: number }[] },
|
||||
newValue: PropertyTypes,
|
||||
delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
update(currentState: { values: { value: string, timestamp: number }[] }, newValue: PropertyTypes, delta: CollapsedDelta) {
|
||||
if (typeof newValue === 'string') {
|
||||
// Check if this value already exists (avoid duplicates)
|
||||
const exists = currentState.values.some(v => v.value === newValue);
|
||||
@ -305,10 +201,7 @@ export class ConcatenationPlugin implements ResolverPlugin<{ values: { value: st
|
||||
return currentState;
|
||||
}
|
||||
|
||||
resolve(
|
||||
state: { values: { value: string, timestamp: number }[] },
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
): PropertyTypes {
|
||||
resolve(state: { values: { value: string, timestamp: number }[] }): PropertyTypes {
|
||||
return state.values.map(v => v.value).join(this.separator);
|
||||
}
|
||||
}
|
||||
@ -321,32 +214,24 @@ export class MajorityVotePlugin implements ResolverPlugin<{ votes: Map<PropertyT
|
||||
return { votes: new Map() };
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { votes: Map<PropertyTypes, number> },
|
||||
newValue: PropertyTypes,
|
||||
_delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
const count = (currentState.votes.get(newValue) || 0) + 1;
|
||||
currentState.votes.set(newValue, count);
|
||||
update(currentState: { votes: Map<PropertyTypes, number> }, newValue: PropertyTypes, _delta: CollapsedDelta) {
|
||||
const currentCount = currentState.votes.get(newValue) || 0;
|
||||
currentState.votes.set(newValue, currentCount + 1);
|
||||
return currentState;
|
||||
}
|
||||
|
||||
resolve(
|
||||
state: { votes: Map<PropertyTypes, number> },
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
): PropertyTypes {
|
||||
resolve(state: { votes: Map<PropertyTypes, number> }): PropertyTypes {
|
||||
let maxVotes = 0;
|
||||
let result: PropertyTypes = '';
|
||||
let winner: PropertyTypes = '';
|
||||
|
||||
for (const [value, count] of state.votes.entries()) {
|
||||
if (count > maxVotes) {
|
||||
maxVotes = count;
|
||||
result = value;
|
||||
for (const [value, votes] of state.votes.entries()) {
|
||||
if (votes > maxVotes) {
|
||||
maxVotes = votes;
|
||||
winner = value;
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
return winner;
|
||||
}
|
||||
}
|
||||
|
||||
@ -358,12 +243,7 @@ export class MinPlugin implements ResolverPlugin<{ min?: number }> {
|
||||
return {};
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { min?: number },
|
||||
newValue: PropertyTypes,
|
||||
_delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
update(currentState: { min?: number }, newValue: PropertyTypes, _delta: CollapsedDelta) {
|
||||
if (typeof newValue === 'number') {
|
||||
if (currentState.min === undefined || newValue < currentState.min) {
|
||||
return { min: newValue };
|
||||
@ -372,10 +252,7 @@ export class MinPlugin implements ResolverPlugin<{ min?: number }> {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
resolve(
|
||||
state: { min?: number },
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
): PropertyTypes | undefined {
|
||||
resolve(state: { min?: number }): PropertyTypes | undefined {
|
||||
return state.min;
|
||||
}
|
||||
}
|
||||
@ -387,12 +264,7 @@ export class MaxPlugin implements ResolverPlugin<{ max?: number }> {
|
||||
return {};
|
||||
}
|
||||
|
||||
update(
|
||||
currentState: { max?: number },
|
||||
newValue: PropertyTypes,
|
||||
_delta: CollapsedDelta,
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
) {
|
||||
update(currentState: { max?: number }, newValue: PropertyTypes, _delta: CollapsedDelta) {
|
||||
if (typeof newValue === 'number') {
|
||||
if (currentState.max === undefined || newValue > currentState.max) {
|
||||
return { max: newValue };
|
||||
@ -401,10 +273,7 @@ export class MaxPlugin implements ResolverPlugin<{ max?: number }> {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
resolve(
|
||||
state: { max?: number },
|
||||
_context?: { entityState: Record<string, unknown>, resolvedValues: Record<string, PropertyTypes> }
|
||||
): PropertyTypes | undefined {
|
||||
resolve(state: { max?: number }): PropertyTypes | undefined {
|
||||
return state.max;
|
||||
}
|
||||
}
|
@ -1,4 +1,3 @@
|
||||
export * from './aggregation-resolvers';
|
||||
export * from './custom-resolvers';
|
||||
export * from './timestamp-resolvers';
|
||||
export * from './relationship-graph';
|
@ -1,288 +0,0 @@
|
||||
import { CollapsedDelta, Lossless, LosslessViewOne } from "../lossless";
|
||||
import { PropertyTypes } from "../../core/types";
|
||||
import { Lossy } from "../lossy";
|
||||
import Debug from 'debug';
|
||||
import { CustomResolver, LastWriteWinsPlugin } from "./custom-resolvers";
|
||||
|
||||
const debug = Debug('rz:resolver:relationship-graph');
|
||||
const trace = Debug('rz:resolver:relationship-graph:trace');
|
||||
trace.enabled = true; // Always enable trace for now
|
||||
|
||||
/**
|
||||
* Represents a single relationship between entities
|
||||
*/
|
||||
export interface Relationship {
|
||||
id: string;
|
||||
type: string;
|
||||
source: string;
|
||||
target: string;
|
||||
properties: Record<string, PropertyTypes>;
|
||||
deltas: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Tracks relationships from a single entity's perspective
|
||||
*/
|
||||
interface EntityRelationships {
|
||||
outbound: Map<string, Relationship>; // relationshipId -> Relationship
|
||||
inbound: Map<string, Relationship>; // relationshipId -> Relationship
|
||||
}
|
||||
|
||||
/**
|
||||
* The accumulator that maintains the complete relationship graph state
|
||||
*/
|
||||
interface RelationshipGraphAccumulator {
|
||||
// Maps entity ID to its relationships
|
||||
entities: Map<string, EntityRelationships>;
|
||||
|
||||
// Maps relationship ID to the relationship data
|
||||
relationships: Map<string, Relationship>;
|
||||
|
||||
// For quick lookups by relationship type
|
||||
relationshipsByType: Map<string, Set<string>>; // type -> Set<relationshipId>
|
||||
|
||||
// Timestamp of the most recent update
|
||||
lastUpdated?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* The public view of relationships for a single entity
|
||||
*/
|
||||
export type RelationshipView = {
|
||||
outbound: Relationship[];
|
||||
inbound: Relationship[];
|
||||
};
|
||||
|
||||
/**
|
||||
* A resolver that builds a relationship graph from a LosslessViewMany
|
||||
*/
|
||||
export class RelationshipGraphResolver extends Lossy<RelationshipGraphAccumulator, RelationshipGraphAccumulator> {
|
||||
private relData: CustomResolver;
|
||||
constructor(lossless: Lossless) {
|
||||
super(lossless);
|
||||
|
||||
this.relData = new CustomResolver(lossless, {
|
||||
relationships: new LastWriteWinsPlugin(),
|
||||
});
|
||||
}
|
||||
/**
|
||||
* Initialize a new accumulator
|
||||
*/
|
||||
protected createRelationshipGraphAccumulator(): RelationshipGraphAccumulator {
|
||||
debug('Creating new relationship graph accumulator');
|
||||
const accumulator = {
|
||||
entities: new Map(),
|
||||
relationships: new Map(),
|
||||
relationshipsByType: new Map(),
|
||||
lastUpdated: Date.now()
|
||||
};
|
||||
trace('Created accumulator:', accumulator);
|
||||
return accumulator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the accumulator with a view
|
||||
*/
|
||||
initializer(view: LosslessViewOne): RelationshipGraphAccumulator {
|
||||
debug('Initializing relationship graph for view:', view.id);
|
||||
const graph = this.createRelationshipGraphAccumulator();
|
||||
|
||||
this.relData.initializer(view);
|
||||
|
||||
trace('Initialized graph state:', {
|
||||
entities: Array.from(graph.entities.keys()),
|
||||
relationships: Array.from(graph.relationships.keys()),
|
||||
relationshipTypes: Array.from(graph.relationshipsByType.keys())
|
||||
});
|
||||
|
||||
return graph;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a view and update the accumulator
|
||||
*/
|
||||
reducer(graph: RelationshipGraphAccumulator, view: LosslessViewOne): RelationshipGraphAccumulator {
|
||||
debug(`Processing view ${view.id} in reducer`);
|
||||
trace('View details:', {
|
||||
id: view.id,
|
||||
propertyCount: Object.keys(view.propertyDeltas).length,
|
||||
properties: Object.keys(view.propertyDeltas)
|
||||
});
|
||||
|
||||
// if (!isRelationshipEntity) {
|
||||
// trace(`Skipping non-relationship delta: ${view.id}`);
|
||||
// return graph;
|
||||
// }
|
||||
|
||||
// Ensure entity exists in the graph
|
||||
if (!graph.entities.has(view.id)) {
|
||||
trace(`Adding new entity in reducer: ${view.id}`);
|
||||
graph.entities.set(view.id, {
|
||||
outbound: new Map(),
|
||||
inbound: new Map()
|
||||
});
|
||||
}
|
||||
|
||||
// Process relationship properties
|
||||
for (const [property, deltas] of Object.entries(view.propertyDeltas)) {
|
||||
trace(`Processing property: ${property} with ${deltas.length} deltas`);
|
||||
|
||||
// Skip non-relationship properties
|
||||
if (!property.startsWith('_rel_')) {
|
||||
trace(`Skipping non-relationship property: ${property}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const delta of deltas) {
|
||||
this.processRelationshipDelta(graph, delta);
|
||||
}
|
||||
}
|
||||
|
||||
graph.lastUpdated = Date.now();
|
||||
return graph;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the accumulator into a final result.
|
||||
* For now, we just return the accumulator as is.
|
||||
*/
|
||||
resolver(graph: RelationshipGraphAccumulator): RelationshipGraphAccumulator {
|
||||
debug('Resolving relationship graph');
|
||||
trace('Graph state at resolution:', {
|
||||
entities: Array.from(graph.entities.keys()),
|
||||
relationships: Array.from(graph.relationships.keys()),
|
||||
relationshipTypes: Array.from(graph.relationshipsByType.entries()).map(([type, ids]) => ({
|
||||
type,
|
||||
count: ids.size
|
||||
}))
|
||||
});
|
||||
return graph;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a single relationship delta
|
||||
*/
|
||||
private processRelationshipDelta(graph: RelationshipGraphAccumulator, delta: CollapsedDelta): void {
|
||||
debug('Processing relationship delta:', delta.id);
|
||||
trace('Delta details:', delta);
|
||||
|
||||
// Extract relationship metadata from the delta
|
||||
const relProps = this.extractRelationshipProperties(delta);
|
||||
if (!relProps) {
|
||||
debug('No relationship properties found in delta:', delta.id);
|
||||
return;
|
||||
}
|
||||
|
||||
trace('Extracted relationship properties:', relProps);
|
||||
|
||||
const { type, sourceId, targetId, relId, properties } = relProps;
|
||||
|
||||
debug(`Processing relationship ${relId} of type ${type} from ${sourceId} to ${targetId}`);
|
||||
|
||||
// Ensure source and target entities exist in the graph
|
||||
if (!graph.entities.has(sourceId)) {
|
||||
trace(`Adding source entity: ${sourceId}`);
|
||||
graph.entities.set(sourceId, { outbound: new Map(), inbound: new Map() });
|
||||
}
|
||||
if (!graph.entities.has(targetId)) {
|
||||
trace(`Adding target entity: ${targetId}`);
|
||||
graph.entities.set(targetId, { outbound: new Map(), inbound: new Map() });
|
||||
}
|
||||
|
||||
// Get or create the relationship
|
||||
let relationship = graph.relationships.get(relId);
|
||||
|
||||
if (!relationship) {
|
||||
debug(`Creating new relationship: ${relId} (${type})`);
|
||||
// Create new relationship
|
||||
relationship = {
|
||||
id: relId,
|
||||
type,
|
||||
source: sourceId,
|
||||
target: targetId,
|
||||
properties: { ...properties },
|
||||
deltas: [delta.id]
|
||||
};
|
||||
|
||||
// Add to relationships map
|
||||
graph.relationships.set(relId, relationship);
|
||||
trace(`Added relationship ${relId} to relationships map`);
|
||||
|
||||
// Add to relationships by type index
|
||||
if (!graph.relationshipsByType.has(type)) {
|
||||
trace(`Creating new relationship type index: ${type}`);
|
||||
graph.relationshipsByType.set(type, new Set());
|
||||
}
|
||||
graph.relationshipsByType.get(type)?.add(relId);
|
||||
trace(`Added relationship ${relId} to type index: ${type}`);
|
||||
|
||||
// Update entity relationships
|
||||
const sourceEntity = graph.entities.get(sourceId)!;
|
||||
const targetEntity = graph.entities.get(targetId)!;
|
||||
|
||||
sourceEntity.outbound.set(relId, relationship);
|
||||
targetEntity.inbound.set(relId, relationship);
|
||||
|
||||
trace('Updated entity relationships:', {
|
||||
sourceOutbound: Array.from(sourceEntity.outbound.keys()),
|
||||
targetInbound: Array.from(targetEntity.inbound.keys())
|
||||
});
|
||||
} else {
|
||||
debug(`Updating existing relationship: ${relId}`);
|
||||
// TODO: Conflict resolution e.g. using TimestampResolver
|
||||
relationship.properties = { ...relationship.properties, ...properties };
|
||||
|
||||
// Track this delta if not already present
|
||||
if (!relationship.deltas.includes(delta.id)) {
|
||||
relationship.deltas.push(delta.id);
|
||||
trace(`Added delta ${delta.id} to relationship ${relId}`);
|
||||
} else {
|
||||
trace(`Delta ${delta.id} already tracked for relationship ${relId}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract relationship properties from a delta
|
||||
*/
|
||||
private extractRelationshipProperties(delta: CollapsedDelta): {
|
||||
relId: string;
|
||||
sourceId: string;
|
||||
targetId: string;
|
||||
type: string;
|
||||
properties: Record<string, PropertyTypes>;
|
||||
} | null {
|
||||
// Get all pointers that are part of this relationship
|
||||
const relPointers = delta.pointers.filter(p =>
|
||||
p._rel_source || p._rel_target || p._rel_type
|
||||
);
|
||||
|
||||
if (relPointers.length === 0) return null;
|
||||
|
||||
// Find the relationship metadata
|
||||
const sourcePtr = relPointers.find(p => p._rel_source);
|
||||
const targetPtr = relPointers.find(p => p._rel_target);
|
||||
const typePtr = relPointers.find(p => p._rel_type);
|
||||
|
||||
if (!sourcePtr || !targetPtr || !typePtr) return null;
|
||||
|
||||
const relId = delta.id; // Use delta ID as relationship ID
|
||||
const sourceId = sourcePtr._rel_source as string;
|
||||
const targetId = targetPtr._rel_target as string;
|
||||
const type = typePtr._rel_type as string;
|
||||
|
||||
// Extract other properties (non-special _rel_ pointers)
|
||||
const properties: Record<string, PropertyTypes> = {};
|
||||
|
||||
for (const ptr of delta.pointers) {
|
||||
for (const [key, value] of Object.entries(ptr)) {
|
||||
if (key.startsWith('_rel_') && !['_rel_source', '_rel_target', '_rel_type'].includes(key)) {
|
||||
const propName = key.substring(5); // Remove '_rel_' prefix
|
||||
properties[propName] = value as PropertyTypes;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { relId, sourceId, targetId, type, properties };
|
||||
}
|
||||
}
|
Loading…
x
Reference in New Issue
Block a user