Define output room event type

federationsender/consumers/roomserver.go

@@ -110,9 +110,6 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 // processMessage updates the list of currently joined hosts in the room
 // and then sends the event to the hosts that were joined before the event.
 func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) error {
-	if ore.Historical {
-		return nil
-	}
 
 	addsJoinedHosts, err := joinedHostsFromEvents(gomatrixserverlib.UnwrapEventHeaders(ore.AddsState()))
 	if err != nil {
@@ -143,7 +140,7 @@ func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) err
 		return nil
 	}
 
-	if ore.SendAsServer == api.DoNotSendToOtherServers {
+	if ore.SendAsServer == api.DoNotSendToOtherServers || ore.Type == api.OutputRoomState {
 		// Ignore event that we don't need to send anywhere.
 		return nil
 	}

roomserver/api/output.go

@@ -63,6 +63,17 @@ type OutputEvent struct {
 	RedactedEvent *OutputRedactedEvent `json:"redacted_event,omitempty"`
 }
 
+// Type of the OutputNewRoomEvent.
+type OutputRoomEventType int
+
+const (
+	// The event is a timeline event and likely just happened.
+	OutputRoomTimeline OutputRoomEventType = iota
+
+	// The event is a state event and quite possibly happened in the past.
+	OutputRoomState
+)
+
 // An OutputNewRoomEvent is written when the roomserver receives a new event.
 // It contains the full matrix room event and enough information for a
 // consumer to construct the current state of the room and the state before the
@@ -75,9 +86,9 @@ type OutputEvent struct {
 type OutputNewRoomEvent struct {
 	// The Event.
 	Event gomatrixserverlib.HeaderedEvent `json:"event"`
-	// Is the event historical? If so, then downstream components should not treat the
-	// event as if it just arrived.
-	Historical bool `json:"historical"`
+	// Is the event a timeline event or a state event? Defaults to timeline
+	// if not specified.
+	Type OutputRoomEventType `json:"type"`
 	// The latest events in the room after this event.
 	// This can be used to set the prev events for new events in the room.
 	// This also can be used to get the full current state after this event.

roomserver/internal/input/input_latest_events.go

@@ -307,9 +307,14 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
 		latestEventIDs[i] = u.latest[i].EventID
 	}
 
+	var outputType api.OutputRoomEventType
+	if u.isHistorical {
+		outputType = api.OutputRoomState
+	}
+
 	ore := api.OutputNewRoomEvent{
 		Event:           u.event.Headered(u.roomInfo.RoomVersion),
-		Historical:      u.isHistorical,
+		Type:            outputType,
 		LastSentEventID: u.lastEventIDSent,
 		LatestEventIDs:  latestEventIDs,
 		TransactionID:   u.transactionID,

syncapi/consumers/roomserver.go

@@ -143,7 +143,7 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
 		}
 	}
 
-	if msg.Historical {
+	if msg.Type == api.OutputRoomState {
 		return nil
 	}