refactor: simplify relationship graph implementation

- Implement RelationshipGraphResolver extending Lossy
- Add type-safe accumulator for relationship graph state
- Update tests and documentation
This commit is contained in:
Lentil Hoffman 2025-06-22 18:39:53 -05:00
parent 76e00f38af
commit 73d2bf23f5
Signed by: lentil
GPG Key ID: 0F5B99F3F4D0C087
45 changed files with 570 additions and 211 deletions

View File

@ -1,11 +1,11 @@
import { createOrchestrator, type NodeConfig } from '../../src/orchestration';
import type { NodeHandle } from '../../src/orchestration/types';
// Increase test timeout to 30 seconds
jest.setTimeout(30000);
describe('Run (Orchestrated)', () => {
const orchestrator = createOrchestrator('in-memory');
let nodeHandle: any;
let nodeHandle: NodeHandle;
let apiUrl: string;
beforeAll(async () => {
@ -14,7 +14,7 @@ describe('Run (Orchestrated)', () => {
id: 'app-001',
};
nodeHandle = await orchestrator.startNode(config);
apiUrl = nodeHandle.getApiUrl();
apiUrl = nodeHandle.getApiUrl?.() || 'http://localhost:3000'; // Default URL if getApiUrl is not available
}, 60000); // Increase timeout to 60s for this hook
afterAll(async () => {

View File

@ -1,16 +1,11 @@
import Docker from 'dockerode';
import { describe, it, beforeAll, afterAll, expect, jest } from '@jest/globals';
import Debug from 'debug';
const debug = Debug('rz:test:docker-orchestrator-v2');
import { createOrchestrator } from '../../src/orchestration';
import type { NodeOrchestrator, NodeConfig, NodeHandle, NodeStatus } from '../../src/orchestration';
import type { NodeConfig, NodeHandle, NodeStatus } from '../../src/orchestration';
import { DockerOrchestrator, createOrchestrator } from '../../src/orchestration';
import { ImageManager } from '../../src/orchestration/docker-orchestrator/managers/image-manager';
import Debug from 'debug';
const debug = Debug('rz:test:docker-orchestrator-v2');
// Extend the NodeOrchestrator type to include the docker client for DockerOrchestrator
interface DockerOrchestrator extends NodeOrchestrator {
docker: Docker;
}
// Extended interface to include additional properties that might be present in the implementation
interface ExtendedNodeStatus extends Omit<NodeStatus, 'network'> {
@ -33,12 +28,8 @@ jest.setTimeout(300000);
describe('Docker Orchestrator', () => {
let docker: Docker;
let orchestrator: DockerOrchestrator;
let node: NodeHandle | null = null;
let node2: NodeHandle | null = null;
let nodeConfig: NodeConfig;
let node2Config: NodeConfig;
let nodePort: number;
let node2Port: number;
beforeAll(async () => {
debug('Setting up Docker client and orchestrator...');
@ -59,7 +50,10 @@ describe('Docker Orchestrator', () => {
}
// Initialize the orchestrator with the Docker client and test image
orchestrator = createOrchestrator('docker') as DockerOrchestrator;
orchestrator = createOrchestrator('docker', {
docker,
image: 'rhizome-node-test',
}) as DockerOrchestrator;
debug('Docker orchestrator initialized');
// Create a basic node config for testing with unique network ID
@ -89,6 +83,9 @@ describe('Docker Orchestrator', () => {
debug('All test cleanups completed');
}, 120000); // 2 minute timeout for afterAll
/**
* ! Note that this test fails if the build fails
*/
it('should start and stop a node', async () => {
debug('Starting test: should start and stop a node');
@ -183,7 +180,7 @@ describe('Docker Orchestrator', () => {
}
// Get container info using ContainerManager
const container = await (orchestrator as any).containerManager.getContainer(status.containerId);
const container = await orchestrator.containerManager.getContainer(status.containerId);
if (!container) {
throw new Error('Container not found');
}
@ -358,11 +355,10 @@ describe('Docker Orchestrator', () => {
// Start first node
debug('Starting node 1...');
node1 = await orchestrator.startNode(node1Config);
debug(`Node 1 started with ID: ${node1.id}`);
debug(`Node 1 started with ID: ${node1.id} apiUrl: ${node1.getApiUrl?.()}`);
// Get node 1's status and API URL
const status1 = await node1.status() as ExtendedNodeStatus;
const node1ApiUrl = node1.getApiUrl?.();
// Update node 2's config with node 1's actual address if available
if (status1.network?.address && node2Config.network) {
@ -373,11 +369,10 @@ describe('Docker Orchestrator', () => {
// Start second node
debug('Starting node 2...');
node2 = await orchestrator.startNode(node2Config);
debug(`Node 2 started with ID: ${node2.id}`);
debug(`Node 2 started with ID: ${node2.id} apiUrl: ${node2.getApiUrl?.()}`);
// Get node 2's status
const status2 = await node2.status() as ExtendedNodeStatus;
const node2ApiUrl = node2.getApiUrl?.();
// Verify both nodes are running
expect(status1).toBeDefined();

View File

@ -1,16 +1,16 @@
import { LosslessViewOne } from '../src/views/lossless';
import { LosslessViewOne } from '../../src/views/lossless';
import {
SchemaBuilder,
PrimitiveSchemas,
ReferenceSchemas,
ArraySchemas,
ObjectSchema
} from '../src/schema';
import { DefaultSchemaRegistry } from '../src/schema';
import { CommonSchemas } from '../util/schemas';
import { TypedCollectionImpl, SchemaValidationError } from '../src/collections';
import { RhizomeNode } from '../src/node';
import { createDelta } from '../src/core/delta-builder';
} from '../../src/schema';
import { DefaultSchemaRegistry } from '../../src/schema';
import { CommonSchemas } from '../../util/schemas';
import { TypedCollectionImpl, SchemaValidationError } from '../../src/collections';
import { RhizomeNode } from '../../src/node';
import { createDelta } from '../../src/core/delta-builder';
describe('Schema System', () => {
let schemaRegistry: DefaultSchemaRegistry;

View File

@ -1,7 +1,6 @@
import { MemoryDeltaStorage, LevelDBDeltaStorage, StorageFactory } from '../src/storage';
import { createDelta } from '../src/core/delta-builder';
import { Delta } from '../src/core';
import { DeltaQueryStorage } from '../src/storage/interface';
import { MemoryDeltaStorage, LevelDBDeltaStorage, StorageFactory } from '../../src/storage';
import { createDelta } from '../../src/core/delta-builder';
import { DeltaQueryStorage } from '../../src/storage/interface';
describe('Delta Storage', () => {
const testDeltas = [

View File

@ -1,15 +0,0 @@
// Set up environment variables for tests
// DEBUG handling examples:
// npm test // will set DEBUG=rz:* by default
// NO_DEBUG=true npm test // will not set DEBUG
// DEBUG=other npm test // will set DEBUG=other
if (!process.env.DEBUG && !process.env.NO_DEBUG) {
// process.env.DEBUG = 'rz:*';
}
// This is a placeholder test to satisfy Jest's requirement for at least one test
describe('Test Setup', () => {
it('should pass', () => {
expect(true).toBe(true);
});
});

View File

@ -1,4 +1,4 @@
import { createDelta } from '../src/core/delta-builder';
import { createDelta } from '../../src/core/delta-builder';
import {
RhizomeNode,
Lossless,
@ -7,7 +7,7 @@ import {
LastWriteWinsPlugin,
MajorityVotePlugin,
TimestampResolver
} from "../src";
} from "../../src";
describe('Concurrent Write Scenarios', () => {
let node: RhizomeNode;

View File

@ -9,13 +9,13 @@
*/
import Debug from 'debug';
import { RhizomeNode } from '../src/node';
import { RhizomeNode } from '../../src/node';
import { createDelta } from '../../src/core';
import { DefaultSchemaRegistry } from '../../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas, ArraySchemas } from '../../src/schema';
import { TypedCollectionImpl } from '../../src/collections';
const debug = Debug('rz:test:nested-resolution-performance');
import { Delta, createDelta } from '../src/core';
import { DefaultSchemaRegistry } from '../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas, ArraySchemas } from '../src/schema';
import { TypedCollectionImpl } from '../src/collections';
describe('Nested Object Resolution Performance', () => {
let node: RhizomeNode;

View File

@ -4,8 +4,8 @@
* to the original deltas with all pointer relationships preserved.
*/
import { RhizomeNode } from '../src/node';
import { createDelta } from '../src/core/delta-builder';
import { RhizomeNode } from '../../../src/node';
import { createDelta } from '../../../src/core/delta-builder';
describe('Lossless View Compose/Decompose', () => {
let node: RhizomeNode;

View File

@ -1,8 +1,8 @@
import { createDelta } from '../src/core/delta-builder';
import { DeltaV1, DeltaV2 } from '../src/core/delta';
import { Lossless } from '../src/views/lossless';
import { RhizomeNode } from '../src/node';
import { TimestampResolver } from '../src/views/resolvers/timestamp-resolvers';
import { createDelta } from '../../../../src/core/delta-builder';
import { DeltaV1, DeltaV2 } from '../../../../src/core/delta';
import { Lossless } from '../../../../src/views/lossless';
import { RhizomeNode } from '../../../../src/node';
import { TimestampResolver } from '../../../../src/views/resolvers/timestamp-resolvers';
describe('DeltaBuilder', () => {
const creator = 'creator-123';

View File

@ -6,7 +6,7 @@ import {
InvalidPointerError,
validateDeltaNetworkImageV1,
validateDeltaNetworkImageV2
} from "../src";
} from "../../../../src";
describe("Delta Validation", () => {
describe("Invalid Delta Formats", () => {

View File

@ -1,5 +1,5 @@
import { createDelta } from '../src/core/delta-builder';
import {DeltaV1, DeltaV2} from "../src";
import { createDelta } from '../../../../src/core/delta-builder';
import {DeltaV1, DeltaV2} from "../../../../src";
describe("Delta", () => {
it("can convert DeltaV1 to DeltaV2", () => {

View File

@ -1,5 +1,5 @@
import { jsonToAst } from '../src/util/json-ast';
import { JsonNode } from '../src/util/json-ast/types';
import { jsonToAst } from '../../../src/util/json-ast';
import { JsonNode } from '../../../src/util/json-ast/types';
describe('jsonToAst', () => {
it('should handle primitive values', () => {

View File

@ -1,8 +1,8 @@
import Debug from 'debug';
import { createDelta } from '../src/core/delta-builder';
import { NegationHelper } from '../src/features';
import { RhizomeNode } from '../src/node';
import { Lossless } from '../src/views';
import { createDelta } from '../../../src/core/delta-builder';
import { NegationHelper } from '../../../src/features';
import { RhizomeNode } from '../../../src/node';
import { Lossless } from '../../../src/views';
const debug = Debug('rz:negation:test');

View File

@ -1,4 +1,4 @@
import {parseAddressList, PeerAddress} from '../src/network/peers';
import {parseAddressList, PeerAddress} from '../../../src/network/peers';
describe('PeerAddress', () => {
it('toString()', () => {

View File

@ -1,7 +1,7 @@
import { createDelta } from '../src/core/delta-builder';
import { Lossless } from '../src/views';
import { RhizomeNode } from '../src/node';
import { DeltaFilter } from '../src/core';
import { createDelta } from '../../../src/core/delta-builder';
import { Lossless } from '../../../src/views';
import { RhizomeNode } from '../../../src/node';
import { DeltaFilter } from '../../../src/core';
describe('Transactions', () => {
let node: RhizomeNode;
@ -211,7 +211,7 @@ describe('Transactions', () => {
const updateEvents: Array<{ entityId: string, deltaIds: string[] }> = [];
// Listen for update events
lossless.eventStream.on('updated', (entityId, deltaIds) => {
lossless.eventStream.on('updated', (entityId: string, deltaIds: string[]) => {
updateEvents.push({ entityId, deltaIds });
});
@ -302,7 +302,7 @@ describe('Transactions', () => {
.buildV1();
const updateEvents: string[] = [];
lossless.eventStream.on('updated', (entityId) => {
lossless.eventStream.on('updated', (entityId: string) => {
updateEvents.push(entityId);
});

View File

@ -1,10 +1,10 @@
import { QueryEngine } from '../src/query';
import { Lossless } from '../src/views';
import { DefaultSchemaRegistry } from '../src/schema';
import { SchemaBuilder, PrimitiveSchemas } from '../src/schema';
import { CommonSchemas } from '../util/schemas';
import { createDelta } from '../src/core/delta-builder';
import { RhizomeNode } from '../src/node';
import { QueryEngine } from '../../../src/query';
import { Lossless } from '../../../src/views';
import { DefaultSchemaRegistry } from '../../../src/schema';
import { SchemaBuilder, PrimitiveSchemas } from '../../../src/schema';
import { CommonSchemas } from '../../../util/schemas';
import { createDelta } from '../../../src/core/delta-builder';
import { RhizomeNode } from '../../../src/node';
describe('Query Engine', () => {
let queryEngine: QueryEngine;

View File

@ -1,7 +1,7 @@
import {Delta, DeltaFilter, DeltaV2} from '../src/core';
import {Lossless} from '../src/views';
import {RhizomeNode} from '../src/node';
import {createDelta} from '../src/core/delta-builder';
import {DeltaFilter} from '../../../src/core';
import {Lossless} from '../../../src/views';
import {RhizomeNode} from '../../../src/node';
import {createDelta} from '../../../src/core/delta-builder';
describe('Lossless', () => {
const node = new RhizomeNode();

View File

@ -1,14 +1,11 @@
import Debug from 'debug';
import {
PointerTarget,
lastValueFromDeltas,
valueFromCollapsedDelta,
Lossless,
LosslessViewOne,
Lossy,
RhizomeNode
} from "../src";
import { createDelta } from "../src/core/delta-builder";
import { PointerTarget } from "../../../src/core/delta";
import { Lossless, LosslessViewOne } from "../../../src/views/lossless";
import { Lossy } from "../../../src/views/lossy";
import { RhizomeNode } from "../../../src/node";
import { valueFromCollapsedDelta } from "../../../src/views/resolvers/aggregation-resolvers";
import { lastValueFromDeltas } from "../../../src/views/resolvers/timestamp-resolvers";
import { createDelta } from "../../../src/core/delta-builder";
const debug = Debug('test:lossy');
type Role = {

View File

@ -4,11 +4,11 @@
* between multiple entities and includes scalar values.
*/
import { RhizomeNode } from '../src/node';
import { createDelta } from '../src/core/delta-builder';
import { DefaultSchemaRegistry } from '../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas, SchemaAppliedViewWithNesting } from '../src/schema';
import { TypedCollectionImpl } from '../src/collections';
import { RhizomeNode } from '../../../src/node';
import { createDelta } from '../../../src/core/delta-builder';
import { DefaultSchemaRegistry } from '../../../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas, SchemaAppliedViewWithNesting } from '../../../src/schema';
import { TypedCollectionImpl } from '../../../src/collections';
describe('Multi-Pointer Delta Resolution', () => {
let node: RhizomeNode;

View File

@ -9,12 +9,12 @@
* - Deep nesting scenarios
*/
import { RhizomeNode } from '../src/node';
import { DefaultSchemaRegistry } from '../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas } from '../src/schema';
import { CommonSchemas } from '../util/schemas';
import { TypedCollectionImpl } from '../src/collections';
import { createDelta } from '../src/core/delta-builder';
import { RhizomeNode } from '../../../src/node';
import { DefaultSchemaRegistry } from '../../../src/schema';
import { SchemaBuilder, PrimitiveSchemas, ReferenceSchemas } from '../../../src/schema';
import { CommonSchemas } from '../../../util/schemas';
import { TypedCollectionImpl } from '../../../src/collections';
import { createDelta } from '../../../src/core/delta-builder';
describe('Nested Object Resolution', () => {
let node: RhizomeNode;

View File

@ -8,9 +8,8 @@ import {
AverageResolver,
CountResolver,
AggregationType
} from "../src";
import { createDelta } from "../src/core/delta-builder";
import { Delta } from "../src/core/delta";
} from "../../../../src";
import { createDelta } from "../../../../src/core/delta-builder";
describe('Aggregation Resolvers', () => {
let node: RhizomeNode;

View File

@ -1,7 +1,6 @@
import {
RhizomeNode,
Lossless,
Delta,
CustomResolver,
ResolverPlugin,
LastWriteWinsPlugin,
@ -13,7 +12,7 @@ import {
PropertyTypes,
CollapsedDelta,
createDelta
} from "../src";
} from "../../../../src";
describe('Custom Resolvers', () => {
let node: RhizomeNode;

View File

@ -1,7 +1,7 @@
import Debug from "debug";
import { createDelta } from '../src/core/delta-builder';
import {Lossless, RhizomeNode} from "../src";
import {TimestampResolver} from "../src/views/resolvers/timestamp-resolvers";
import { createDelta } from '../../../../src/core/delta-builder';
import { Lossless, RhizomeNode } from '../../../../src';
import { TimestampResolver } from '../../../../src/views/resolvers/timestamp-resolvers';
const debug = Debug('test:last-write-wins');
// This was initially written to test a LastWriteWins resolver, but that has been

View File

@ -6,8 +6,8 @@ import {
DeltaIdTimestampResolver,
HostIdTimestampResolver,
LexicographicTimestampResolver
} from "../src";
import { createDelta } from "../src/core/delta-builder";
} from "../../../../src";
import { createDelta } from "../../../../src/core/delta-builder";
describe('Timestamp Resolvers', () => {
let node: RhizomeNode;

132
docs/classes.md Normal file
View File

@ -0,0 +1,132 @@
# Rhizome Node Class Diagram
This document provides an overview of the main classes in the Rhizome Node system and their relationships.
```mermaid
classDiagram
%% Core Classes
class RhizomeNode {
-config: RhizomeNodeConfig
-pubSub: PubSub
-requestReply: RequestReply
-httpServer: HttpServer
-deltaStream: DeltaStream
-lossless: Lossless
-peers: Peers
-queryEngine: QueryEngine
-storageQueryEngine: StorageQueryEngine
-schemaRegistry: SchemaRegistry
-deltaStorage: DeltaStorage
}
class Delta {
+id: DeltaID
+timeCreated: Timestamp
+host: HostID
+creator: CreatorID
+pointers: PointerV1[]
}
class Lossless {
-domainEntities: Map<DomainEntityID, LosslessEntity>
-transactions: Transactions
+view(ids: DomainEntityID[]): LosslessViewMany
+compose(ids: DomainEntityID[]): LosslessViewMany
}
class QueryEngine {
-lossless: Lossless
-schemaRegistry: SchemaRegistry
+query(schemaId: SchemaID, filter?: JsonLogic): Promise<SchemaAppliedViewWithNesting[]>
}
class StorageQueryEngine {
-storage: DeltaQueryStorage
-schemaRegistry: SchemaRegistry
+query(schemaId: SchemaID, filter?: JsonLogic): Promise<StorageQueryResult>
}
class SchemaRegistry {
+register(schema: ObjectSchema): void
+get(id: SchemaID): ObjectSchema | undefined
+list(): ObjectSchema[]
}
class DeltaStream {
-deltas: Delta[]
+receiveDelta(delta: Delta): void
+ingestAll(): void
}
class DockerOrchestrator {
-containerManager: ContainerManager
-networkManager: NetworkManager
-resourceManager: ResourceManager
-statusManager: StatusManager
+startNode(config: NodeConfig): Promise<NodeHandle>
+stopNode(nodeId: string): Promise<void>
}
%% Relationships
RhizomeNode --> DeltaStream
RhizomeNode --> Lossless
RhizomeNode --> QueryEngine
RhizomeNode --> StorageQueryEngine
RhizomeNode --> SchemaRegistry
RhizomeNode --> DeltaStorage
Lossless --> Transactions
Lossless --> LosslessEntity
QueryEngine --> SchemaRegistry
QueryEngine --> Lossless
StorageQueryEngine --> DeltaStorage
StorageQueryEngine --> SchemaRegistry
DeltaStream --> Delta
Lossless --> Delta
DockerOrchestrator --> ContainerManager
DockerOrchestrator --> NetworkManager
DockerOrchestrator --> ResourceManager
DockerOrchestrator --> StatusManager
```
## Key Components
1. **RhizomeNode**: The main entry point that coordinates all other components
- Manages the node's lifecycle and configuration
- Coordinates between different subsystems
2. **Delta**: The fundamental data unit
- Represents atomic changes in the system
- Contains pointers to entities and their properties
3. **Lossless**: Manages the lossless view of data
- Maintains the complete history of deltas
- Provides methods to view and compose entity states
4. **QueryEngine**: Handles data queries
- Supports filtering with JSON Logic
- Works with the schema system for validation
5. **StorageQueryEngine**: Handles storage-level queries
- Interfaces with the underlying storage backend
- Optimized for querying persisted data
6. **SchemaRegistry**: Manages data schemas
- Validates data against schemas
- Supports schema versioning and evolution
7. **DockerOrchestrator**: Manages containerized nodes
- Handles node lifecycle (start/stop)
- Manages networking between nodes
## Data Flow
1. Deltas are received through the DeltaStream
2. Lossless processes and stores these deltas
3. Queries can be made through either QueryEngine (in-memory) or StorageQueryEngine (persisted)
4. The system maintains consistency through the schema system
5. In distributed mode, DockerOrchestrator manages multiple node instances

15
docs/resolvers.md Normal file
View File

@ -0,0 +1,15 @@
# Resolvers (Views)
The workhorse of this system is likely going to be our lossy views.
This is where the computation likely generally occurs.
So, let's talk about how to create a view.
A lossy view initializes from a given lossless view.
The lossless view dispatches events when entity properties are updated.
View semantics are similar to map-reduce, resolvers in Redux, etc.
The key is to identify your accumulator object.
Your algorithm SHOULD be implemented so that the reducer is a pure function.
All state must therefore be stored in the accumulator.

View File

@ -7,7 +7,7 @@
"build:watch": "tsc --watch",
"lint": "eslint",
"test": "jest",
"coverage": "npm run test -- --coverage --coverageDirectory=coverage",
"coverage": "npm run test -- --coverage",
"example-app": "node dist/examples/app.js",
"stop-all": "docker ps -a --filter \"name=^/rhizome-node-\" --format {{.Names}} | xargs -r docker stop",
"build-test-image": "docker build -t rhizome-node-test -f Dockerfile.test ."
@ -16,14 +16,19 @@
"testEnvironment": "node",
"preset": "ts-jest/presets/default",
"roots": [
"./__tests__/"
"<rootDir>/__tests__"
],
"testMatch": [
"**/__tests__/**/*"
"**/*.test.ts",
"**/*.test.tsx"
],
"setupFilesAfterEnv": [
"<rootDir>/__tests__/jest-setup.ts"
"testPathIgnorePatterns": [
"/node_modules/",
"/dist/"
],
"moduleNameMapper": {
"^@/(.*)$": "<rootDir>/src/$1"
},
"transform": {
"^\\.tsx?$": [
"ts-jest",
@ -37,7 +42,16 @@
"verbose": true,
"testEnvironmentOptions": {
"NODE_ENV": "test"
}
},
"collectCoverageFrom": [
"src/**/*.{ts,tsx}",
"util/**/*.{ts,tsx}",
"!**/node_modules/**",
"!**/dist/**",
"!**/__tests__/**",
"!**/*.test.{ts,tsx}"
],
"coverageDirectory": "coverage"
},
"author": "Lentil <lentil@laddhoffman.com>",
"license": "Unlicense",

9
plans/dimensions.md Normal file
View File

@ -0,0 +1,9 @@
# Dimensions
Deltas are a hyperstructure capable of expressing many structures.
We can consider models which we compose by expressing their dimensions, and
relationships between those dimensions. This can enable a robust form of multi-dimensional
composition. I'm thinking in terms of vectors and matrices, maybe quaternions, tensors.
So basically we can use deltas to encode math.

View File

@ -1,5 +1,7 @@
# Housekeeping tasks
- [] Organize tests?
- [] More documentation in docs/
- [] Rename/consolidate, lossless view() and compose() --> composeView()
- [x] Organize tests?
- [ ] More documentation in docs/
- [ ] Rename/consolidate, lossless view() and compose() --> composeView()
- [ ] Rename Lossless to HyperView
- [ ] Rename Lossy to View

View File

@ -4,7 +4,7 @@ import EventEmitter from "node:events";
import {Delta} from "../core/delta";
import {createDelta} from "../core/delta-builder";
import {Entity, EntityProperties} from "../core/entity";
import {ResolvedViewOne} from '../views/resolvers/last-write-wins';
import { ResolvedTimestampedViewOne as ResolvedViewOne } from '../views/resolvers/timestamp-resolvers';
import {RhizomeNode} from "../node";
import {DomainEntityID} from "../core/types";
const debug = Debug('rz:abstract-collection');

View File

@ -4,7 +4,7 @@
// It could then be further extended with e.g. table semantics like filter, sort, join
import {Collection} from '../collections/collection-abstract';
import {ResolvedViewOne} from '../views/resolvers/last-write-wins';
import { ResolvedTimestampedViewOne as ResolvedViewOne } from '../views/resolvers/timestamp-resolvers';
import {TimestampResolver} from '../views/resolvers/timestamp-resolvers';
export class BasicCollection extends Collection<TimestampResolver> {

View File

@ -1,5 +1,5 @@
import {Collection} from "./collection-abstract";
import {ResolvedViewOne} from "../views/resolvers/last-write-wins";
import { ResolvedTimestampedViewOne as ResolvedViewOne } from "../views/resolvers/timestamp-resolvers";
import {TimestampResolver} from "../views/resolvers/timestamp-resolvers";
class RelationalView extends TimestampResolver {

View File

@ -1,6 +1,6 @@
import Debug from 'debug';
import { Collection } from '../collections/collection-abstract';
import { ResolvedViewOne } from '../views/resolvers/last-write-wins';
import { ResolvedTimestampedViewOne as ResolvedViewOne } from '../views/resolvers/timestamp-resolvers';
import { TimestampResolver } from '../views/resolvers/timestamp-resolvers'
import {
ObjectSchema,

View File

@ -26,7 +26,8 @@ export class DockerOrchestrator extends BaseOrchestrator {
private nodeHandles: Map<string, DockerNodeHandle> = new Map();
// Managers
private readonly containerManager: ContainerManager;
// Visible for testing
readonly containerManager: ContainerManager;
private readonly networkManager: NetworkManager;
private readonly resourceManager: ResourceManager;
private readonly statusManager: StatusManager;
@ -225,7 +226,7 @@ export class DockerOrchestrator extends BaseOrchestrator {
/**
* Create network partitions
*/
async partitionNetwork(partitions: NetworkPartition): Promise<void> {
async partitionNetwork(_partitions: NetworkPartition): Promise<void> {
// Implementation for network partitioning
// This is a simplified version - in a real implementation, you would:
// 1. Create separate networks for each partition

View File

@ -37,7 +37,7 @@ export interface NodeConfig {
};
/** Additional configuration options */
[key: string]: any;
[key: string]: unknown;
}
export interface NodeStatus {

View File

@ -5,9 +5,18 @@
import Debug from 'debug';
import {DeltaFilter, DeltaID} from "../core/delta";
import {Lossless, LosslessViewOne} from "./lossless";
import {DomainEntityID} from "../core/types";
import {DomainEntityID, PropertyID, PropertyTypes, ViewMany} from "../core/types";
const debug = Debug('rz:lossy');
type PropertyMap = Record<PropertyID, PropertyTypes>;
export type LossyViewOne<T = PropertyMap> = {
id: DomainEntityID;
properties: T;
};
export type LossyViewMany<T = PropertyMap> = ViewMany<LossyViewOne<T>>;
// We support incremental updates of lossy models.
export abstract class Lossy<Accumulator, Result> {
deltaFilter?: DeltaFilter;

View File

@ -2,7 +2,7 @@ import { EntityProperties } from "../../core/entity";
import { Lossless, LosslessViewOne } from "../lossless";
import { Lossy } from '../lossy';
import { DomainEntityID, PropertyID, ViewMany } from "../../core/types";
import { valueFromCollapsedDelta } from "./last-write-wins";
import { CollapsedDelta } from "../lossless";
export type AggregationType = 'min' | 'max' | 'sum' | 'average' | 'count';
@ -37,6 +37,20 @@ type ResolvedAggregatedViewMany = ViewMany<ResolvedAggregatedViewOne>;
type Accumulator = AggregatedViewMany;
type Result = ResolvedAggregatedViewMany;
// Extract a particular value from a delta's pointers
export function valueFromCollapsedDelta(
key: string,
delta: CollapsedDelta
): string | number | undefined {
for (const pointer of delta.pointers) {
for (const [k, value] of Object.entries(pointer)) {
if (k === key && (typeof value === "string" || typeof value === "number")) {
return value;
}
}
}
}
function aggregateValues(values: number[], type: AggregationType): number {
if (values.length === 0) return 0;

View File

@ -1,4 +1,4 @@
export * from './aggregation-resolvers';
export * from './custom-resolvers';
export * from './last-write-wins';
export * from './timestamp-resolvers';
export * from './timestamp-resolvers';
export * from './relationship-graph';

View File

@ -1,69 +0,0 @@
// import Debug from 'debug';
import {EntityProperties} from "../../core/entity";
import {CollapsedDelta} from "../lossless";
import {DomainEntityID, PropertyID, PropertyTypes, Timestamp, ViewMany} from "../../core/types";
// const debug = Debug('rz:lossy:last-write-wins');
type TimestampedProperty = {
value: PropertyTypes,
timeUpdated: Timestamp
};
type TimestampedProperties = {
[key: PropertyID]: TimestampedProperty
};
export type LossyViewOne<T = TimestampedProperties> = {
id: DomainEntityID;
properties: T;
};
export type LossyViewMany<T = TimestampedProperties> = ViewMany<LossyViewOne<T>>;
export type ResolvedViewOne = LossyViewOne<EntityProperties>;
export type ResolvedViewMany = ViewMany<ResolvedViewOne>;
type Accumulator = LossyViewMany<TimestampedProperties>;
type Result = LossyViewMany<EntityProperties>;
// Extract a particular value from a delta's pointers
export function valueFromCollapsedDelta(
key: string,
delta: CollapsedDelta
): string | number | undefined {
for (const pointer of delta.pointers) {
for (const [k, value] of Object.entries(pointer)) {
if (k === key && (typeof value === "string" || typeof value === "number")) {
return value;
}
}
}
}
// Resolve a value for an entity by last write wins
export function lastValueFromDeltas(
key: string,
deltas?: CollapsedDelta[]
): {
delta?: CollapsedDelta,
value?: string | number,
timeUpdated?: number
} | undefined {
const res: {
delta?: CollapsedDelta,
value?: string | number,
timeUpdated?: number
} = {};
res.timeUpdated = 0;
for (const delta of deltas || []) {
const value = valueFromCollapsedDelta(key, delta);
if (value === undefined) continue;
if (res.timeUpdated && delta.timeCreated < res.timeUpdated) continue;
res.delta = delta;
res.value = value;
res.timeUpdated = delta.timeCreated;
}
return res;
}

View File

@ -0,0 +1,219 @@
import { CollapsedDelta, LosslessViewOne } from "../lossless";
import { PropertyTypes } from "../../core/types";
import { Lossy } from "../lossy";
/**
* 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> {
/**
* Initialize a new accumulator
*/
protected createRelationshipGraphAccumulator(): RelationshipGraphAccumulator {
return {
entities: new Map(),
relationships: new Map(),
relationshipsByType: new Map(),
lastUpdated: Date.now()
};
}
/**
* Initialize the accumulator with a view
*/
initializer(view: LosslessViewOne): RelationshipGraphAccumulator {
const graph = this.createRelationshipGraphAccumulator();
// Initialize entity relationships if they don't exist
if (!graph.entities.has(view.id)) {
graph.entities.set(view.id, {
outbound: new Map(),
inbound: new Map()
});
}
return graph;
}
/**
* Process a view and update the accumulator
*/
reducer(graph: RelationshipGraphAccumulator, view: LosslessViewOne): RelationshipGraphAccumulator {
// Ensure entity exists in the graph
if (!graph.entities.has(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)) {
// Skip non-relationship properties
if (!property.startsWith('_rel_')) 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 {
return graph;
}
/**
* Process a single relationship delta
*/
private processRelationshipDelta(graph: RelationshipGraphAccumulator, delta: CollapsedDelta): void {
// Extract relationship metadata from the delta
const relProps = this.extractRelationshipProperties(delta);
if (!relProps) return;
const { type, sourceId, targetId, relId, properties } = relProps;
// Ensure source and target entities exist in the graph
if (!graph.entities.has(sourceId)) {
graph.entities.set(sourceId, { outbound: new Map(), inbound: new Map() });
}
if (!graph.entities.has(targetId)) {
graph.entities.set(targetId, { outbound: new Map(), inbound: new Map() });
}
// Get or create the relationship
let relationship = graph.relationships.get(relId);
if (!relationship) {
// 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);
// Add to relationships by type index
if (!graph.relationshipsByType.has(type)) {
graph.relationshipsByType.set(type, new Set());
}
graph.relationshipsByType.get(type)?.add(relId);
// 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);
} else {
// Update existing relationship
// 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);
}
}
}
/**
* 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 };
}
}

View File

@ -1,8 +1,19 @@
import { EntityProperties } from "../../core/entity";
import { Lossless, LosslessViewOne } from "../lossless";
import { Lossless, LosslessViewOne, CollapsedDelta } from "../lossless";
import { Lossy } from '../lossy';
import { DomainEntityID, PropertyID, PropertyTypes, Timestamp, ViewMany } from "../../core/types";
import { valueFromCollapsedDelta } from "./last-write-wins";
import { valueFromCollapsedDelta } from "./aggregation-resolvers";
// Moved here from last-write-wins.ts before removing that file
export type TimestampedProperty = {
value: PropertyTypes,
timeUpdated: Timestamp
};
// Moved here from last-write-wins.ts before removing that file
export type TimestampedProperties = {
[key: PropertyID]: TimestampedProperty
};
export type TieBreakingStrategy = 'creator-id' | 'delta-id' | 'host-id' | 'lexicographic';
@ -152,4 +163,32 @@ export class LexicographicTimestampResolver extends TimestampResolver {
constructor(lossless: Lossless) {
super(lossless, 'lexicographic');
}
}
// Resolve a value for an entity by last write wins
export function lastValueFromDeltas(
key: string,
deltas?: CollapsedDelta[]
): {
delta?: CollapsedDelta,
value?: string | number,
timeUpdated?: number
} | undefined {
const res: {
delta?: CollapsedDelta,
value?: string | number,
timeUpdated?: number
} = {};
res.timeUpdated = 0;
for (const delta of deltas || []) {
const value = valueFromCollapsedDelta(key, delta);
if (value === undefined) continue;
if (res.timeUpdated && delta.timeCreated < res.timeUpdated) continue;
res.delta = delta;
res.value = value;
res.timeUpdated = delta.timeCreated;
}
return res;
}