// Copyright 2020 The Matrix.org Foundation C.I.C. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // Package msc2836 'Threading' implements https://github.com/matrix-org/matrix-doc/pull/2836 package msc2836 import ( "bytes" "context" "crypto/sha256" "encoding/json" "fmt" "io" "net/http" "sort" "strings" "time" "github.com/matrix-org/dendrite/clientapi/jsonerror" fs "github.com/matrix-org/dendrite/federationsender/api" "github.com/matrix-org/dendrite/internal/hooks" "github.com/matrix-org/dendrite/internal/httputil" roomserver "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/setup" userapi "github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" ) const ( constRelType = "m.reference" ) type EventRelationshipRequest struct { EventID string `json:"event_id"` RoomID string `json:"room_id"` MaxDepth int `json:"max_depth"` MaxBreadth int `json:"max_breadth"` Limit int `json:"limit"` DepthFirst bool `json:"depth_first"` RecentFirst bool `json:"recent_first"` IncludeParent bool `json:"include_parent"` IncludeChildren bool `json:"include_children"` Direction string `json:"direction"` Batch string `json:"batch"` } func NewEventRelationshipRequest(body io.Reader) (*EventRelationshipRequest, error) { var relation EventRelationshipRequest relation.Defaults() if err := json.NewDecoder(body).Decode(&relation); err != nil { return nil, err } return &relation, nil } func (r *EventRelationshipRequest) Defaults() { r.Limit = 100 r.MaxBreadth = 10 r.MaxDepth = 3 r.DepthFirst = false r.RecentFirst = true r.IncludeParent = false r.IncludeChildren = false r.Direction = "down" } type EventRelationshipResponse struct { Events []gomatrixserverlib.ClientEvent `json:"events"` NextBatch string `json:"next_batch"` Limited bool `json:"limited"` } func toClientResponse(res *gomatrixserverlib.MSC2836EventRelationshipsResponse) *EventRelationshipResponse { out := &EventRelationshipResponse{ Events: gomatrixserverlib.ToClientEvents(res.Events, gomatrixserverlib.FormatAll), Limited: res.Limited, NextBatch: res.NextBatch, } return out } // Enable this MSC func Enable( base *setup.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI, userAPI userapi.UserInternalAPI, keyRing gomatrixserverlib.JSONVerifier, ) error { db, err := NewDatabase(&base.Cfg.MSCs.Database) if err != nil { return fmt.Errorf("Cannot enable MSC2836: %w", err) } hooks.Enable() hooks.Attach(hooks.KindNewEventPersisted, func(headeredEvent interface{}) { he := headeredEvent.(*gomatrixserverlib.HeaderedEvent) hookErr := db.StoreRelation(context.Background(), he) if hookErr != nil { util.GetLogger(context.Background()).WithError(hookErr).WithField("event_id", he.EventID()).Error( "failed to StoreRelation", ) } // we need to update child metadata here as well as after doing remote /event_relationships requests // so we catch child metadata originating from /send transactions hookErr = db.UpdateChildMetadata(context.Background(), he) if hookErr != nil { util.GetLogger(context.Background()).WithError(err).WithField("event_id", he.EventID()).Warn( "failed to update child metadata for event", ) } }) base.PublicClientAPIMux.Handle("/unstable/event_relationships", httputil.MakeAuthAPI("eventRelationships", userAPI, eventRelationshipHandler(db, rsAPI, fsAPI)), ).Methods(http.MethodPost, http.MethodOptions) base.PublicFederationAPIMux.Handle("/unstable/event_relationships", httputil.MakeExternalAPI( "msc2836_event_relationships", func(req *http.Request) util.JSONResponse { fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest( req, time.Now(), base.Cfg.Global.ServerName, keyRing, ) if fedReq == nil { return errResp } return federatedEventRelationship(req.Context(), fedReq, db, rsAPI, fsAPI) }, )).Methods(http.MethodPost, http.MethodOptions) return nil } type reqCtx struct { ctx context.Context rsAPI roomserver.RoomserverInternalAPI db Database req *EventRelationshipRequest userID string roomVersion gomatrixserverlib.RoomVersion // federated request args isFederatedRequest bool serverName gomatrixserverlib.ServerName fsAPI fs.FederationSenderInternalAPI } func eventRelationshipHandler(db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI) func(*http.Request, *userapi.Device) util.JSONResponse { return func(req *http.Request, device *userapi.Device) util.JSONResponse { relation, err := NewEventRelationshipRequest(req.Body) if err != nil { util.GetLogger(req.Context()).WithError(err).Error("failed to decode HTTP request as JSON") return util.JSONResponse{ Code: 400, JSON: jsonerror.BadJSON(fmt.Sprintf("invalid json: %s", err)), } } rc := reqCtx{ ctx: req.Context(), req: relation, userID: device.UserID, rsAPI: rsAPI, fsAPI: fsAPI, isFederatedRequest: false, db: db, } res, resErr := rc.process() if resErr != nil { return *resErr } return util.JSONResponse{ Code: 200, JSON: toClientResponse(res), } } } func federatedEventRelationship( ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI, ) util.JSONResponse { relation, err := NewEventRelationshipRequest(bytes.NewBuffer(fedReq.Content())) if err != nil { util.GetLogger(ctx).WithError(err).Error("failed to decode HTTP request as JSON") return util.JSONResponse{ Code: 400, JSON: jsonerror.BadJSON(fmt.Sprintf("invalid json: %s", err)), } } rc := reqCtx{ ctx: ctx, req: relation, rsAPI: rsAPI, db: db, // federation args isFederatedRequest: true, fsAPI: fsAPI, serverName: fedReq.Origin(), } res, resErr := rc.process() if resErr != nil { return *resErr } // add auth chain information requiredAuthEventsSet := make(map[string]bool) var requiredAuthEvents []string for _, ev := range res.Events { for _, a := range ev.AuthEventIDs() { if requiredAuthEventsSet[a] { continue } requiredAuthEvents = append(requiredAuthEvents, a) requiredAuthEventsSet[a] = true } } var queryRes roomserver.QueryAuthChainResponse err = rsAPI.QueryAuthChain(ctx, &roomserver.QueryAuthChainRequest{ EventIDs: requiredAuthEvents, }, &queryRes) if err != nil { // they may already have the auth events so don't fail this request util.GetLogger(ctx).WithError(err).Error("Failed to QueryAuthChain") } res.AuthChain = make([]*gomatrixserverlib.Event, len(queryRes.AuthChain)) for i := range queryRes.AuthChain { res.AuthChain[i] = queryRes.AuthChain[i].Unwrap() } return util.JSONResponse{ Code: 200, JSON: res, } } func (rc *reqCtx) process() (*gomatrixserverlib.MSC2836EventRelationshipsResponse, *util.JSONResponse) { var res gomatrixserverlib.MSC2836EventRelationshipsResponse var returnEvents []*gomatrixserverlib.HeaderedEvent // Can the user see (according to history visibility) event_id? If no, reject the request, else continue. event := rc.getLocalEvent(rc.req.EventID) if event == nil { event = rc.fetchUnknownEvent(rc.req.EventID, rc.req.RoomID) } if rc.req.RoomID == "" && event != nil { rc.req.RoomID = event.RoomID() } if event == nil || !rc.authorisedToSeeEvent(event) { return nil, &util.JSONResponse{ Code: 403, JSON: jsonerror.Forbidden("Event does not exist or you are not authorised to see it"), } } rc.roomVersion = event.Version() // Retrieve the event. Add it to response array. returnEvents = append(returnEvents, event) if rc.req.IncludeParent { if parentEvent := rc.includeParent(event); parentEvent != nil { returnEvents = append(returnEvents, parentEvent) } } if rc.req.IncludeChildren { remaining := rc.req.Limit - len(returnEvents) if remaining > 0 { children, resErr := rc.includeChildren(rc.db, event.EventID(), remaining, rc.req.RecentFirst) if resErr != nil { return nil, resErr } returnEvents = append(returnEvents, children...) } } remaining := rc.req.Limit - len(returnEvents) var walkLimited bool if remaining > 0 { included := make(map[string]bool, len(returnEvents)) for _, ev := range returnEvents { included[ev.EventID()] = true } var events []*gomatrixserverlib.HeaderedEvent events, walkLimited = walkThread( rc.ctx, rc.db, rc, included, remaining, ) returnEvents = append(returnEvents, events...) } res.Events = make([]*gomatrixserverlib.Event, len(returnEvents)) for i, ev := range returnEvents { // for each event, extract the children_count | hash and add it as unsigned data. rc.addChildMetadata(ev) res.Events[i] = ev.Unwrap() } res.Limited = remaining == 0 || walkLimited return &res, nil } // fetchUnknownEvent retrieves an unknown event from the room specified. This server must // be joined to the room in question. This has the side effect of injecting surround threaded // events into the roomserver. func (rc *reqCtx) fetchUnknownEvent(eventID, roomID string) *gomatrixserverlib.HeaderedEvent { if rc.isFederatedRequest || roomID == "" { // we don't do fed hits for fed requests, and we can't ask servers without a room ID! return nil } logger := util.GetLogger(rc.ctx).WithField("room_id", roomID) // if they supplied a room_id, check the room exists. var queryVerRes roomserver.QueryRoomVersionForRoomResponse err := rc.rsAPI.QueryRoomVersionForRoom(rc.ctx, &roomserver.QueryRoomVersionForRoomRequest{ RoomID: roomID, }, &queryVerRes) if err != nil { logger.WithError(err).Warn("failed to query room version for room, does this room exist?") return nil } // check the user is joined to that room var queryMemRes roomserver.QueryMembershipForUserResponse err = rc.rsAPI.QueryMembershipForUser(rc.ctx, &roomserver.QueryMembershipForUserRequest{ RoomID: roomID, UserID: rc.userID, }, &queryMemRes) if err != nil { logger.WithError(err).Warn("failed to query membership for user in room") return nil } if !queryMemRes.IsInRoom { return nil } // ask one of the servers in the room for the event var queryRes fs.QueryJoinedHostServerNamesInRoomResponse err = rc.fsAPI.QueryJoinedHostServerNamesInRoom(rc.ctx, &fs.QueryJoinedHostServerNamesInRoomRequest{ RoomID: roomID, }, &queryRes) if err != nil { logger.WithError(err).Error("failed to QueryJoinedHostServerNamesInRoom") return nil } // query up to 5 servers serversToQuery := queryRes.ServerNames if len(serversToQuery) > 5 { serversToQuery = serversToQuery[:5] } // fetch the event, along with some of the surrounding thread (if it's threaded) and the auth chain. // Inject the response into the roomserver to remember the event across multiple calls and to set // unexplored flags correctly. for _, srv := range serversToQuery { res, err := rc.MSC2836EventRelationships(eventID, srv, queryVerRes.RoomVersion) if err != nil { continue } rc.injectResponseToRoomserver(res) for _, ev := range res.Events { if ev.EventID() == eventID { return ev.Headered(ev.Version()) } } } logger.WithField("servers", serversToQuery).Warn("failed to query event relationships") return nil } // If include_parent: true and there is a valid m.relationship field in the event, // retrieve the referenced event. Apply history visibility check to that event and if it passes, add it to the response array. func (rc *reqCtx) includeParent(childEvent *gomatrixserverlib.HeaderedEvent) (parent *gomatrixserverlib.HeaderedEvent) { parentID, _, _ := parentChildEventIDs(childEvent) if parentID == "" { return nil } return rc.lookForEvent(parentID) } // If include_children: true, lookup all events which have event_id as an m.relationship // Apply history visibility checks to all these events and add the ones which pass into the response array, // honouring the recent_first flag and the limit. func (rc *reqCtx) includeChildren(db Database, parentID string, limit int, recentFirst bool) ([]*gomatrixserverlib.HeaderedEvent, *util.JSONResponse) { if rc.hasUnexploredChildren(parentID) { // we need to do a remote request to pull in the children as we are missing them locally. serversToQuery := rc.getServersForEventID(parentID) var result *gomatrixserverlib.MSC2836EventRelationshipsResponse for _, srv := range serversToQuery { res, err := rc.fsAPI.MSC2836EventRelationships(rc.ctx, srv, gomatrixserverlib.MSC2836EventRelationshipsRequest{ EventID: parentID, Direction: "down", Limit: 100, MaxBreadth: -1, MaxDepth: 1, // we just want the children from this parent RecentFirst: true, }, rc.roomVersion) if err != nil { util.GetLogger(rc.ctx).WithError(err).WithField("server", srv).Error("includeChildren: failed to call MSC2836EventRelationships") } else { result = &res break } } if result != nil { rc.injectResponseToRoomserver(result) } // fallthrough to pull these new events from the DB } children, err := db.ChildrenForParent(rc.ctx, parentID, constRelType, recentFirst) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("failed to get ChildrenForParent") resErr := jsonerror.InternalServerError() return nil, &resErr } var childEvents []*gomatrixserverlib.HeaderedEvent for _, child := range children { childEvent := rc.lookForEvent(child.EventID) if childEvent != nil { childEvents = append(childEvents, childEvent) } } if len(childEvents) > limit { return childEvents[:limit], nil } return childEvents, nil } // Begin to walk the thread DAG in the direction specified, either depth or breadth first according to the depth_first flag, // honouring the limit, max_depth and max_breadth values according to the following rules func walkThread( ctx context.Context, db Database, rc *reqCtx, included map[string]bool, limit int, ) ([]*gomatrixserverlib.HeaderedEvent, bool) { var result []*gomatrixserverlib.HeaderedEvent eventWalker := walker{ ctx: ctx, req: rc.req, db: db, fn: func(wi *walkInfo) bool { // If already processed event, skip. if included[wi.EventID] { return false } // If the response array is >= limit, stop. if len(result) >= limit { return true } // Process the event. // if event is not found, use remoteEventRelationships to explore that part of the thread remotely. // This will probably be easiest if the event relationships response is directly pumped into the database // so the next walk will do the right thing. This requires those events to be authed and likely injected as // outliers into the roomserver DB, which will de-dupe appropriately. event := rc.lookForEvent(wi.EventID) if event != nil { result = append(result, event) } included[wi.EventID] = true return false }, } limited, err := eventWalker.WalkFrom(rc.req.EventID) if err != nil { util.GetLogger(ctx).WithError(err).Errorf("Failed to WalkFrom %s", rc.req.EventID) } return result, limited } // MSC2836EventRelationships performs an /event_relationships request to a remote server func (rc *reqCtx) MSC2836EventRelationships(eventID string, srv gomatrixserverlib.ServerName, ver gomatrixserverlib.RoomVersion) (*gomatrixserverlib.MSC2836EventRelationshipsResponse, error) { res, err := rc.fsAPI.MSC2836EventRelationships(rc.ctx, srv, gomatrixserverlib.MSC2836EventRelationshipsRequest{ EventID: eventID, DepthFirst: rc.req.DepthFirst, Direction: rc.req.Direction, Limit: rc.req.Limit, MaxBreadth: rc.req.MaxBreadth, MaxDepth: rc.req.MaxDepth, RecentFirst: rc.req.RecentFirst, }, ver) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("Failed to call MSC2836EventRelationships") return nil, err } return &res, nil } // authorisedToSeeEvent checks that the user or server is allowed to see this event. Returns true if allowed to // see this request. This only needs to be done once per room at present as we just check for joined status. func (rc *reqCtx) authorisedToSeeEvent(event *gomatrixserverlib.HeaderedEvent) bool { if rc.isFederatedRequest { // make sure the server is in this room var res fs.QueryJoinedHostServerNamesInRoomResponse err := rc.fsAPI.QueryJoinedHostServerNamesInRoom(rc.ctx, &fs.QueryJoinedHostServerNamesInRoomRequest{ RoomID: event.RoomID(), }, &res) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("authorisedToSeeEvent: failed to QueryJoinedHostServerNamesInRoom") return false } for _, srv := range res.ServerNames { if srv == rc.serverName { return true } } return false } // make sure the user is in this room // Allow events if the member is in the room // TODO: This does not honour history_visibility // TODO: This does not honour m.room.create content var queryMembershipRes roomserver.QueryMembershipForUserResponse err := rc.rsAPI.QueryMembershipForUser(rc.ctx, &roomserver.QueryMembershipForUserRequest{ RoomID: event.RoomID(), UserID: rc.userID, }, &queryMembershipRes) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("authorisedToSeeEvent: failed to QueryMembershipForUser") return false } return queryMembershipRes.IsInRoom } func (rc *reqCtx) getServersForEventID(eventID string) []gomatrixserverlib.ServerName { if rc.req.RoomID == "" { util.GetLogger(rc.ctx).WithField("event_id", eventID).Error( "getServersForEventID: event exists in unknown room", ) return nil } if rc.roomVersion == "" { util.GetLogger(rc.ctx).WithField("event_id", eventID).Errorf( "getServersForEventID: event exists in %s with unknown room version", rc.req.RoomID, ) return nil } var queryRes fs.QueryJoinedHostServerNamesInRoomResponse err := rc.fsAPI.QueryJoinedHostServerNamesInRoom(rc.ctx, &fs.QueryJoinedHostServerNamesInRoomRequest{ RoomID: rc.req.RoomID, }, &queryRes) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("getServersForEventID: failed to QueryJoinedHostServerNamesInRoom") return nil } // query up to 5 servers serversToQuery := queryRes.ServerNames if len(serversToQuery) > 5 { serversToQuery = serversToQuery[:5] } return serversToQuery } func (rc *reqCtx) remoteEventRelationships(eventID string) *gomatrixserverlib.MSC2836EventRelationshipsResponse { if rc.isFederatedRequest { return nil // we don't query remote servers for remote requests } serversToQuery := rc.getServersForEventID(eventID) var res *gomatrixserverlib.MSC2836EventRelationshipsResponse var err error for _, srv := range serversToQuery { res, err = rc.MSC2836EventRelationships(eventID, srv, rc.roomVersion) if err != nil { util.GetLogger(rc.ctx).WithError(err).WithField("server", srv).Error("remoteEventRelationships: failed to call MSC2836EventRelationships") } else { break } } return res } // lookForEvent returns the event for the event ID given, by trying to query remote servers // if the event ID is unknown via /event_relationships. func (rc *reqCtx) lookForEvent(eventID string) *gomatrixserverlib.HeaderedEvent { event := rc.getLocalEvent(eventID) if event == nil { queryRes := rc.remoteEventRelationships(eventID) if queryRes != nil { // inject all the events into the roomserver then return the event in question rc.injectResponseToRoomserver(queryRes) for _, ev := range queryRes.Events { if ev.EventID() == eventID && rc.req.RoomID == ev.RoomID() { return ev.Headered(ev.Version()) } } } } else if rc.hasUnexploredChildren(eventID) { // we have the local event but we may need to do a remote hit anyway if we are exploring the thread and have unknown children. // If we don't do this then we risk never fetching the children. queryRes := rc.remoteEventRelationships(eventID) if queryRes != nil { rc.injectResponseToRoomserver(queryRes) err := rc.db.MarkChildrenExplored(context.Background(), eventID) if err != nil { util.GetLogger(rc.ctx).WithError(err).Warnf("failed to mark children of %s as explored", eventID) } } } if rc.req.RoomID == event.RoomID() { return event } return nil } func (rc *reqCtx) getLocalEvent(eventID string) *gomatrixserverlib.HeaderedEvent { var queryEventsRes roomserver.QueryEventsByIDResponse err := rc.rsAPI.QueryEventsByID(rc.ctx, &roomserver.QueryEventsByIDRequest{ EventIDs: []string{eventID}, }, &queryEventsRes) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("getLocalEvent: failed to QueryEventsByID") return nil } if len(queryEventsRes.Events) == 0 { util.GetLogger(rc.ctx).WithField("event_id", eventID).Infof("getLocalEvent: event does not exist") return nil // event does not exist } return queryEventsRes.Events[0] } // injectResponseToRoomserver injects the events // into the roomserver as KindOutlier, with auth chains. func (rc *reqCtx) injectResponseToRoomserver(res *gomatrixserverlib.MSC2836EventRelationshipsResponse) { var stateEvents []*gomatrixserverlib.Event var messageEvents []*gomatrixserverlib.Event for _, ev := range res.Events { if ev.StateKey() != nil { stateEvents = append(stateEvents, ev) } else { messageEvents = append(messageEvents, ev) } } respState := gomatrixserverlib.RespState{ AuthEvents: res.AuthChain, StateEvents: stateEvents, } eventsInOrder, err := respState.Events() if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("failed to calculate order to send events in MSC2836EventRelationshipsResponse") return } // everything gets sent as an outlier because auth chain events may be disjoint from the DAG // as may the threaded events. var ires []roomserver.InputRoomEvent for _, outlier := range append(eventsInOrder, messageEvents...) { ires = append(ires, roomserver.InputRoomEvent{ Kind: roomserver.KindOutlier, Event: outlier.Headered(outlier.Version()), AuthEventIDs: outlier.AuthEventIDs(), }) } // we've got the data by this point so use a background context err = roomserver.SendInputRoomEvents(context.Background(), rc.rsAPI, ires) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("failed to inject MSC2836EventRelationshipsResponse into the roomserver") } // update the child count / hash columns for these nodes. We need to do this here because not all events will make it // through to the KindNewEventPersisted hook because the roomserver will ignore duplicates. Duplicates have meaning though // as the `unsigned` field may differ (if the number of children changes). for _, ev := range ires { err = rc.db.UpdateChildMetadata(context.Background(), ev.Event) if err != nil { util.GetLogger(rc.ctx).WithError(err).WithField("event_id", ev.Event.EventID()).Warn("failed to update child metadata for event") } } } func (rc *reqCtx) addChildMetadata(ev *gomatrixserverlib.HeaderedEvent) { count, hash := rc.getChildMetadata(ev.EventID()) if count == 0 { return } err := ev.SetUnsignedField("children_hash", gomatrixserverlib.Base64Bytes(hash)) if err != nil { util.GetLogger(rc.ctx).WithError(err).Warn("Failed to set children_hash") } err = ev.SetUnsignedField("children", map[string]int{ constRelType: count, }) if err != nil { util.GetLogger(rc.ctx).WithError(err).Warn("Failed to set children count") } } func (rc *reqCtx) getChildMetadata(eventID string) (count int, hash []byte) { children, err := rc.db.ChildrenForParent(rc.ctx, eventID, constRelType, false) if err != nil { util.GetLogger(rc.ctx).WithError(err).Warn("Failed to get ChildrenForParent for getting child metadata") return } if len(children) == 0 { return } // sort it lexiographically sort.Slice(children, func(i, j int) bool { return children[i].EventID < children[j].EventID }) // hash it var eventIDs strings.Builder for _, c := range children { _, _ = eventIDs.WriteString(c.EventID) } hashValBytes := sha256.Sum256([]byte(eventIDs.String())) count = len(children) hash = hashValBytes[:] return } // hasUnexploredChildren returns true if this event has unexplored children. // "An event has unexplored children if the `unsigned` child count on the parent does not match // how many children the server believes the parent to have. In addition, if the counts match but // the hashes do not match, then the event is unexplored." func (rc *reqCtx) hasUnexploredChildren(eventID string) bool { if rc.isFederatedRequest { return false // we only explore children for clients, not servers. } // extract largest child count from event eventCount, eventHash, explored, err := rc.db.ChildMetadata(rc.ctx, eventID) if err != nil { util.GetLogger(rc.ctx).WithError(err).WithField("event_id", eventID).Warn( "failed to get ChildMetadata from db", ) return false } // if there are no recorded children then we know we have >= children. // if the event has already been explored (read: we hit /event_relationships successfully) // then don't do it again. We'll only re-do this if we get an even bigger children count, // see Database.UpdateChildMetadata if eventCount == 0 || explored { return false // short-circuit } // calculate child count for event calcCount, calcHash := rc.getChildMetadata(eventID) if eventCount < calcCount { return false // we have more children } else if eventCount > calcCount { return true // the event has more children than we know about } // we have the same count, so a mismatched hash means some children are different return !bytes.Equal(eventHash, calcHash) } type walkInfo struct { eventInfo SiblingNumber int Depth int } type walker struct { ctx context.Context req *EventRelationshipRequest db Database fn func(wi *walkInfo) bool // callback invoked for each event walked, return true to terminate the walk } // WalkFrom the event ID given func (w *walker) WalkFrom(eventID string) (limited bool, err error) { children, err := w.childrenForParent(eventID) if err != nil { util.GetLogger(w.ctx).WithError(err).Error("WalkFrom() childrenForParent failed, cannot walk") return false, err } var next *walkInfo toWalk := w.addChildren(nil, children, 1) next, toWalk = w.nextChild(toWalk) for next != nil { stop := w.fn(next) if stop { return true, nil } // find the children's children children, err = w.childrenForParent(next.EventID) if err != nil { util.GetLogger(w.ctx).WithError(err).Error("WalkFrom() childrenForParent failed, cannot walk") return false, err } toWalk = w.addChildren(toWalk, children, next.Depth+1) next, toWalk = w.nextChild(toWalk) } return false, nil } // addChildren adds an event's children to the to walk data structure func (w *walker) addChildren(toWalk []walkInfo, children []eventInfo, depthOfChildren int) []walkInfo { // Check what number child this event is (ordered by recent_first) compared to its parent, does it exceed (greater than) max_breadth? If yes, skip. if len(children) > w.req.MaxBreadth { children = children[:w.req.MaxBreadth] } // Check how deep the event is compared to event_id, does it exceed (greater than) max_depth? If yes, skip. if depthOfChildren > w.req.MaxDepth { return toWalk } if w.req.DepthFirst { // the slice is a stack so push them in reverse order so we pop them in the correct order // e.g [3,2,1] => [3,2] , 1 => [3] , 2 => [] , 3 for i := len(children) - 1; i >= 0; i-- { toWalk = append(toWalk, walkInfo{ eventInfo: children[i], SiblingNumber: i + 1, // index from 1 Depth: depthOfChildren, }) } } else { // the slice is a queue so push them in normal order to we dequeue them in the correct order // e.g [1,2,3] => 1, [2, 3] => 2 , [3] => 3, [] for i := range children { toWalk = append(toWalk, walkInfo{ eventInfo: children[i], SiblingNumber: i + 1, // index from 1 Depth: depthOfChildren, }) } } return toWalk } func (w *walker) nextChild(toWalk []walkInfo) (*walkInfo, []walkInfo) { if len(toWalk) == 0 { return nil, nil } var child walkInfo if w.req.DepthFirst { // toWalk is a stack so pop the child off child, toWalk = toWalk[len(toWalk)-1], toWalk[:len(toWalk)-1] return &child, toWalk } // toWalk is a queue so shift the child off child, toWalk = toWalk[0], toWalk[1:] return &child, toWalk } // childrenForParent returns the children events for this event ID, honouring the direction: up|down flags // meaning this can actually be returning the parent for the event instead of the children. func (w *walker) childrenForParent(eventID string) ([]eventInfo, error) { if w.req.Direction == "down" { return w.db.ChildrenForParent(w.ctx, eventID, constRelType, w.req.RecentFirst) } // find the event to pull out the parent ei, err := w.db.ParentForChild(w.ctx, eventID, constRelType) if err != nil { return nil, err } if ei != nil { return []eventInfo{*ei}, nil } return nil, nil }