diff --git a/shims/synapse/shim.py b/shims/synapse/shim.py
index 1f17512..933531d 100644
--- a/shims/synapse/shim.py
+++ b/shims/synapse/shim.py
@@ -1,42 +1,52 @@
#!/usr/bin/env python
import asyncio
import json
+import logging
import uuid
-from typing import Dict, Optional, Sequence
+from typing import Collection, Dict, Iterable, List, Optional, Sequence, Set
from pydantic import BaseModel
from websockets.asyncio.server import serve
+
+from twisted.internet import defer
+from synapse.api.room_versions import RoomVersions
+from synapse.state.v2 import resolve_events_with_store
+from synapse.events import EventBase, make_event_from_dict
+
+# logging.basicConfig(level=logging.DEBUG)
+
class WebSocketMessage(BaseModel):
type: str
id: str
error: Optional[str] = None
data: dict
+room_ver = RoomVersions.V10 # TODO: parameterise
+
class Connection:
+ event_map: Dict[str, EventBase] = {}
def __init__(self, ws):
self.ws = ws
self.outstanding_requests = {}
# Array>
- async def resolve_state(self, id: str, state):
+ async def resolve_state(self, id: str, room_id: str, state):
print(f"resolve_state: {id}")
- for x in state:
- for _, event_id in x.items():
- print(f" get_event {event_id}")
- ev = await self.get_event(event_id)
- print(f" get_event {event_id} obtained. type={ev["type"]}")
+ r = await resolve_events_with_store(FakeClock(),room_id, room_ver, state, event_map=None, state_res_store=self)
print(f"resolve_state: {id} responding")
+ # convert tuple keys to strings
+ r = {json.dumps(k):v for k,v in r.items()}
await self.ws.send(json.dumps({
"id": id,
"type": "resolve_state",
"data": {
- "result": state, # echo it back for now
+ "result": r,
}
}))
return []
- async def get_event(self, event_id: str) -> dict:
+ async def get_event(self, event_id: str) -> EventBase:
id = str(uuid.uuid4())
print(f" get_event {event_id} -> {id}")
loop = asyncio.get_running_loop()
@@ -50,7 +60,74 @@ async def get_event(self, event_id: str) -> dict:
}
}))
await fut
- return self.outstanding_requests[id].result()
+ ev_dict = self.outstanding_requests[id].result()
+ ev_dict.pop("event_id") # wire format shouldn't have this, but tardis includes it.
+ return make_event_from_dict(ev_dict, room_version=room_ver)
+
+ async def get_events(
+ self, event_ids: Collection[str], allow_rejected: bool = False
+ ) -> "defer.Deferred[Dict[str, EventBase]]":
+ """Get events from the database
+
+ Args:
+ event_ids: The event_ids of the events to fetch
+ allow_rejected: If True return rejected events.
+
+ Returns:
+ Dict from event_id to event.
+ """
+ result = {}
+ for event_id in event_ids:
+ print(f" get_event {event_id}")
+ ev = await self.get_event(event_id)
+ print(f" get_event {event_id} obtained. type={ev["type"]}")
+ result[event_id] = ev
+
+ return result
+
+ async def _get_auth_chain(self, event_ids: Iterable[str]) -> List[str]:
+ """Gets the full auth chain for a set of events (including rejected
+ events).
+
+ Includes the given event IDs in the result.
+
+ Note that:
+ 1. All events must be state events.
+ 2. For v1 rooms this may not have the full auth chain in the
+ presence of rejected events
+
+ Args:
+ event_ids: The event IDs of the events to fetch the auth
+ chain for. Must be state events.
+ Returns:
+ List of event IDs of the auth chain.
+ """
+
+ # Simple DFS for auth chain
+ result = set()
+ stack = list(event_ids)
+ while stack:
+ event_id = stack.pop()
+ if event_id in result:
+ continue
+
+ result.add(event_id)
+
+ event = self.event_map.get(event_id, None)
+ if event is None:
+ event = await self.get_event(event_id)
+ self.event_map[event_id] = event
+ for aid in event.auth_event_ids():
+ stack.append(aid)
+
+ return list(result)
+
+ async def get_auth_chain_difference(
+ self, room_id: str, auth_sets: List[Set[str]]
+ ) -> "defer.Deferred[Set[str]]":
+ chains = [frozenset(await self._get_auth_chain(a)) for a in auth_sets]
+ common = set(chains[0]).intersection(*chains[1:])
+ return set(chains[0]).union(*chains[1:]) - common
async def handler(websocket):
@@ -68,12 +145,22 @@ async def handler(websocket):
# call get_event which needs more WS messages, so we cannot block the processing
# of incoming WS messages. When resolve_state concludes, it will send the response,
# hence why we pass in the id here so it can pair it up.
- asyncio.create_task(c.resolve_state(wsm.id, wsm.data["state"]))
+ asyncio.create_task(c.resolve_state(wsm.id, wsm.data["room_id"], wsm.data["state"]))
else:
print(f"unknown type: {wsm.type}")
except Exception as err:
print(f"recv error {err}")
+# Synapse specifics below
+# -----------------------
+
+class FakeClock:
+ def sleep(self, msec: float) -> "defer.Deferred[None]":
+ return defer.succeed(None)
+
+
+
+
async def main():
print("Listening on 0.0.0.0:1234")
async with serve(handler, "0.0.0.0", 1234):
@@ -81,4 +168,4 @@ async def main():
if __name__ == '__main__':
- asyncio.run(main())
\ No newline at end of file
+ asyncio.run(main())
diff --git a/src/index.ts b/src/index.ts
index a67c1da..f6a5df8 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -20,6 +20,9 @@ class Dag {
startEventId: string;
step: number;
eventIdFileOrdering: string[];
+ eventIdWinners: Set;
+ eventIdLosers: Set;
+ renderEvents: Record;
constructor() {
this.cache = Object.create(null);
@@ -35,6 +38,9 @@ class Dag {
this.startEventId = "";
this.step = 0;
this.eventIdFileOrdering = [];
+ this.eventIdWinners = new Set();
+ this.eventIdLosers = new Set();
+ this.renderEvents = {};
}
async load(file: File) {
const events = await new Promise((resolve: (value: Array) => void, reject) => {
@@ -129,6 +135,7 @@ class Dag {
if (this.collapse) {
renderEvents = this.collapsifier(renderEvents);
}
+ this.renderEvents = renderEvents;
await this.render(this.eventsToCompleteDag(renderEvents));
}
// returns the set of events to render
@@ -188,7 +195,6 @@ class Dag {
// return the first N events in file ordering
async recalculateStep(): Promise> {
const renderEvents = Object.create(null);
- console.log("recalc step", this.eventIdFileOrdering, this.step, this.cache);
for (let i = 0; i < this.step; i++) {
const eventId = this.eventIdFileOrdering[i];
if (!eventId) {
@@ -276,6 +282,7 @@ class Dag {
type: "...",
content: {},
sender: "",
+ room_id: "!",
depth: 0,
};
} else {
@@ -287,6 +294,7 @@ class Dag {
type: "missing",
content: {},
sender: "",
+ room_id: "!",
depth: 0,
};
}
@@ -618,7 +626,15 @@ class Dag {
nodes
.append("circle")
.attr("r", nodeRadius)
- .attr("fill", (n) => colorMap[n.id]);
+ .attr("fill", (n) => {
+ if (this.eventIdLosers.has(n.id)) {
+ return "red";
+ }
+ if (this.eventIdWinners.has(n.id)) {
+ return "green";
+ }
+ return "black";
+ });
// Add text to nodes with border
const getLabel = (d) => {
@@ -733,19 +749,6 @@ document.getElementById("start")!.addEventListener("change", (ev) => {
}
dag = new Dag();
await dag.load(files[0]);
-
- // TODO: remove, just for testing.
- try {
- await transport.connect(resolver);
- const r = await resolver.resolveState(
- Object.keys(dag.cache).map((eventId) => {
- return dag.cache[eventId];
- }),
- );
- console.log("Resolved state:", r);
- } catch (err) {
- console.error("failed to setup WS connection:", err);
- }
},
false,
);
@@ -766,3 +769,25 @@ document.getElementById("stepbwd")!.addEventListener("click", async (ev) => {
dag.stepBackward();
dag.refresh();
});
+document.getElementById("resolve")!.addEventListener("click", async (ev) => {
+ try {
+ await transport.connect(resolver);
+ const r = await resolver.resolveState(
+ Object.keys(dag.renderEvents)
+ .filter((eventId) => {
+ return dag.cache[eventId].state_key != null;
+ })
+ .map((eventId) => {
+ return dag.cache[eventId];
+ }),
+ );
+ console.log("Resolved state:", r);
+ dag.eventIdLosers = new Set(r.lostEventIds);
+ dag.eventIdWinners = new Set(r.wonEventIds);
+ } catch (err) {
+ console.error("failed to setup WS connection:", err);
+ } finally {
+ transport.close();
+ }
+ dag.refresh();
+});
diff --git a/src/state_resolver.test.ts b/src/state_resolver.test.ts
index 484c41c..66713bf 100644
--- a/src/state_resolver.test.ts
+++ b/src/state_resolver.test.ts
@@ -13,6 +13,7 @@ describe("StateResolver", () => {
event_id: "$foo",
sender: "@alice",
depth: 0,
+ room_id: "!foo",
},
$foomember: {
type: "m.room.member",
@@ -23,6 +24,7 @@ describe("StateResolver", () => {
event_id: "$foomember",
sender: "@alice",
depth: 1,
+ room_id: "!foo",
},
$bar: {
type: "m.room.create",
@@ -33,6 +35,7 @@ describe("StateResolver", () => {
event_id: "$bar",
sender: "@alice",
depth: 0,
+ room_id: "!foo",
},
};
@@ -65,12 +68,14 @@ describe("StateResolver", () => {
const fooRequest = outstandingRequests[0];
expect(fooRequest.id).toBeDefined();
expect(fooRequest.data).toEqual({
+ room_id: "!foo",
// biome-ignore lint/complexity/useLiteralKeys: it reads much nicer in IDEs to use this form
state: [{ [`["m.room.create",""]`]: "$foo" }, { [`["m.room.member","@alice"]`]: "$foomember" }],
});
const barRequest = outstandingRequests[1];
expect(barRequest.id).toBeDefined();
expect(barRequest.data).toEqual({
+ room_id: "!foo",
// biome-ignore lint/complexity/useLiteralKeys: it reads much nicer in IDEs to use this form
state: [{ [`["m.room.create",""]`]: "$bar" }],
});
@@ -83,7 +88,8 @@ describe("StateResolver", () => {
sr.onResolveStateResponse(barRequest.id, {
state: [],
// biome-ignore lint/complexity/useLiteralKeys: it reads much nicer in IDEs to use this form
- result: [{ [`["m.room.create",""]`]: "$bar" }],
+ result: { [`["m.room.create",""]`]: "$bar" },
+ room_id: "!foo",
});
const barResult = await promiseBar;
expect(barResolved).toBe(true);
@@ -94,7 +100,8 @@ describe("StateResolver", () => {
sr.onResolveStateResponse(fooRequest.id, {
state: [],
// biome-ignore lint/complexity/useLiteralKeys: it reads much nicer in IDEs to use this form
- result: [{ [`["m.room.create",""]`]: "$foo" }],
+ result: { [`["m.room.create",""]`]: "$foo" },
+ room_id: "!foo",
});
const fooResult = await promiseFoo;
expect(fooResolved).toBe(true);
diff --git a/src/state_resolver.ts b/src/state_resolver.ts
index 046cc7d..89b94af 100644
--- a/src/state_resolver.ts
+++ b/src/state_resolver.ts
@@ -8,6 +8,7 @@ interface MatrixEvent {
depth: number;
prev_events: Array;
auth_events: Array;
+ room_id: string;
// TODO: fix metadata fields
_collapse?: number;
@@ -30,8 +31,9 @@ type StateKeyTuple = string; // JSON encoded array of 2 string elements [type, s
type EventID = string;
interface DataResolveState {
+ room_id: string;
state: Array>;
- result?: Array>;
+ result?: Record;
}
interface DataGetEvent {
event_id: string;
@@ -79,11 +81,13 @@ class StateResolver implements StateResolverReceiver {
// convert events into a form suitable for sending over the wire
const state: Array> = [];
const initialSetOfEventIds = new Set();
+ let roomId = "";
for (const ev of stateEvents) {
state.push({
[`${JSON.stringify([ev.type, ev.state_key])}`]: ev.event_id,
});
initialSetOfEventIds.add(ev.event_id);
+ roomId = ev.room_id;
}
console.log("resolveState", state);
// make an id so we can pair it up when we get the response
@@ -99,16 +103,9 @@ class StateResolver implements StateResolverReceiver {
}
// map the won event IDs
const wonEventIds = new Set(
- resolvedData.result
- .map((tupleToEventId: Record) => {
- for (const tuple in tupleToEventId) {
- return tupleToEventId[tuple];
- }
- console.error(
- "resolveState response has malformed tuple-to-event-id dict:",
- tupleToEventId,
- );
- return "";
+ Object.keys(resolvedData.result)
+ .map((tuple: string): string => {
+ return resolvedData.result![tuple] || "";
})
.values(),
);
@@ -127,6 +124,7 @@ class StateResolver implements StateResolverReceiver {
});
this.sender.sendResolveState(id, {
state: state,
+ room_id: roomId,
});
});