func (r *Inputer) processRoomEvent(
	ctx context.Context,
	input *api.InputRoomEvent,
) (eventID string, err error) {
	// Parse and validate the event JSON
	headered := input.Event
	event := headered.Unwrap()

	// Check that the event passes authentication checks and work out
	// the numeric IDs for the auth events.
	isRejected := false
	authEventNIDs, rejectionErr := helpers.CheckAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
	if rejectionErr != nil {
		logrus.WithError(rejectionErr).WithField("event_id", event.EventID()).WithField("auth_event_ids", input.AuthEventIDs).Error("helpers.CheckAuthEvents failed for event, rejecting event")
		isRejected = true
	}

	var softfail bool
	if input.Kind == api.KindBackfill || input.Kind == api.KindNew {
		// Check that the event passes authentication checks based on the
		// current room state.
		softfail, err = helpers.CheckForSoftFail(ctx, r.DB, headered, input.StateEventIDs)
		if err != nil {
			logrus.WithFields(logrus.Fields{
				"event_id": event.EventID(),
				"type":     event.Type(),
				"room":     event.RoomID(),
			}).WithError(err).Info("Error authing soft-failed event")
		}
	}

	// If we don't have a transaction ID then get one.
	if input.TransactionID != nil {
		tdID := input.TransactionID
		eventID, err = r.DB.GetTransactionEventID(
			ctx, tdID.TransactionID, tdID.SessionID, event.Sender(),
		)
		// On error OR event with the transaction already processed/processesing
		if err != nil || eventID != "" {
			return
		}
	}

	// Store the event.
	_, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs, isRejected)
	if err != nil {
		return "", fmt.Errorf("r.DB.StoreEvent: %w", err)
	}

	// if storing this event results in it being redacted then do so.
	if !isRejected && redactedEventID == event.EventID() {
		r, rerr := eventutil.RedactEvent(redactionEvent, &event)
		if rerr != nil {
			return "", fmt.Errorf("eventutil.RedactEvent: %w", rerr)
		}
		event = *r
	}

	// For outliers we can stop after we've stored the event itself as it
	// doesn't have any associated state to store and we don't need to
	// notify anyone about it.
	if input.Kind == api.KindOutlier {
		logrus.WithFields(logrus.Fields{
			"event_id": event.EventID(),
			"type":     event.Type(),
			"room":     event.RoomID(),
			"sender":   event.Sender(),
		}).Debug("Stored outlier")
		return event.EventID(), nil
	}

	roomInfo, err := r.DB.RoomInfo(ctx, event.RoomID())
	if err != nil {
		return "", fmt.Errorf("r.DB.RoomInfo: %w", err)
	}
	if roomInfo == nil {
		return "", fmt.Errorf("r.DB.RoomInfo missing for room %s", event.RoomID())
	}

	if stateAtEvent.BeforeStateSnapshotNID == 0 {
		// We haven't calculated a state for this event yet.
		// Lets calculate one.
		err = r.calculateAndSetState(ctx, input, *roomInfo, &stateAtEvent, event, isRejected)
		if err != nil {
			return "", fmt.Errorf("r.calculateAndSetState: %w", err)
		}
	}

	// We stop here if the event is rejected: We've stored it but won't update forward extremities or notify anyone about it.
	if isRejected || softfail {
		logrus.WithFields(logrus.Fields{
			"event_id":  event.EventID(),
			"type":      event.Type(),
			"room":      event.RoomID(),
			"soft_fail": softfail,
			"sender":    event.Sender(),
		}).Debug("Stored rejected event")
		return event.EventID(), rejectionErr
	}

	if input.Kind == api.KindRewrite {
		logrus.WithFields(logrus.Fields{
			"event_id": event.EventID(),
			"type":     event.Type(),
			"room":     event.RoomID(),
		}).Debug("Stored rewrite")
		return event.EventID(), nil
	}

	if err = r.updateLatestEvents(
		ctx,                 // context
		roomInfo,            // room info for the room being updated
		stateAtEvent,        // state at event (below)
		event,               // event
		input.SendAsServer,  // send as server
		input.TransactionID, // transaction ID
		input.HasState,      // rewrites state?
	); err != nil {
		return "", fmt.Errorf("r.updateLatestEvents: %w", err)
	}

	// processing this event resulted in an event (which may not be the one we're processing)
	// being redacted. We are guaranteed to have both sides (the redaction/redacted event),
	// so notify downstream components to redact this event - they should have it if they've
	// been tracking our output log.
	if redactedEventID != "" {
		err = r.WriteOutputEvents(event.RoomID(), []api.OutputEvent{
			{
				Type: api.OutputTypeRedactedEvent,
				RedactedEvent: &api.OutputRedactedEvent{
					RedactedEventID: redactedEventID,
					RedactedBecause: redactionEvent.Headered(headered.RoomVersion),
				},
			},
		})
		if err != nil {
			return "", fmt.Errorf("r.WriteOutputEvents: %w", err)
		}
	}

	// Update the extremities of the event graph for the room
	return event.EventID(), nil
}
// processRoomEvent can only be called once at a time
//
// TODO(#375): This should be rewritten to allow concurrent calls. The
// difficulty is in ensuring that we correctly annotate events with the correct
// state deltas when sending to kafka streams
// TODO: Break up function - we should probably do transaction ID checks before calling this.
// nolint:gocyclo
processRoomEvent can only be called once at a time


// processRoomEvent can only be called once at a time // // TODO ( #375 ) : This should be rewritten to allow concurrent calls . The // difficulty is in ensuring that we correctly annotate events with the correct // state deltas when sending to kafka streams // TODO : Break up function - we should probably do transaction ID checks before calling this . // nolint : gocyclo
func (r *Inputer) processRoomEvent(
	ctx context.Context,
	input *api.InputRoomEvent,
) (eventID string, err error) {
	headered := input.Event
	event := headered.Unwrap()
	isRejected := false
	authEventNIDs, rejectionErr := helpers.CheckAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
	if rejectionErr != nil {
		logrus.WithError(rejectionErr).WithField("event_id", event.EventID()).WithField("auth_event_ids", input.AuthEventIDs).Error("helpers.CheckAuthEvents failed for event, rejecting event")
		isRejected = true
	}
	var softfail bool
	if input.Kind == api.KindBackfill || input.Kind == api.KindNew {
		softfail, err = helpers.CheckForSoftFail(ctx, r.DB, headered, input.StateEventIDs)
		if err != nil {
			logrus.WithFields(logrus.Fields{
				"event_id": event.EventID(),
				"type":     event.Type(),
				"room":     event.RoomID(),
			}).WithError(err).Info("Error authing soft-failed event")
		}
	}
	if input.TransactionID != nil {
		tdID := input.TransactionID
		eventID, err = r.DB.GetTransactionEventID(
			ctx, tdID.TransactionID, tdID.SessionID, event.Sender(),
		)
		if err != nil || eventID != "" {
			return
		}
	}
	_, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs, isRejected)
	if err != nil {
		return "", fmt.Errorf("r.DB.StoreEvent: %w", err)
	}
	if !isRejected && redactedEventID == event.EventID() {
		r, rerr := eventutil.RedactEvent(redactionEvent, &event)
		if rerr != nil {
			return "", fmt.Errorf("eventutil.RedactEvent: %w", rerr)
		}
		event = *r
	}
	if input.Kind == api.KindOutlier {
		logrus.WithFields(logrus.Fields{
			"event_id": event.EventID(),
			"type":     event.Type(),
			"room":     event.RoomID(),
			"sender":   event.Sender(),
		}).Debug("Stored outlier")
		return event.EventID(), nil
	}
	roomInfo, err := r.DB.RoomInfo(ctx, event.RoomID())
	if err != nil {
		return "", fmt.Errorf("r.DB.RoomInfo: %w", err)
	}
	if roomInfo == nil {
		return "", fmt.Errorf("r.DB.RoomInfo missing for room %s", event.RoomID())
	}
	if stateAtEvent.BeforeStateSnapshotNID == 0 {
		err = r.calculateAndSetState(ctx, input, *roomInfo, &stateAtEvent, event, isRejected)
		if err != nil {
			return "", fmt.Errorf("r.calculateAndSetState: %w", err)
		}
	}
	if isRejected || softfail {
		logrus.WithFields(logrus.Fields{
			"event_id":  event.EventID(),
			"type":      event.Type(),
			"room":      event.RoomID(),
			"soft_fail": softfail,
			"sender":    event.Sender(),
		}).Debug("Stored rejected event")
		return event.EventID(), rejectionErr
	}
	if input.Kind == api.KindRewrite {
		logrus.WithFields(logrus.Fields{
			"event_id": event.EventID(),
			"type":     event.Type(),
			"room":     event.RoomID(),
		}).Debug("Stored rewrite")
		return event.EventID(), nil
	}
	if err = r.updateLatestEvents(
		ctx,                 
		roomInfo,            
		stateAtEvent,        
		event,               
		input.SendAsServer,  
		input.TransactionID, 
		input.HasState,      
	); err != nil {
		return "", fmt.Errorf("r.updateLatestEvents: %w", err)
	}
	if redactedEventID != "" {
		err = r.WriteOutputEvents(event.RoomID(), []api.OutputEvent{
			{
				Type: api.OutputTypeRedactedEvent,
				RedactedEvent: &api.OutputRedactedEvent{
					RedactedEventID: redactedEventID,
					RedactedBecause: redactionEvent.Headered(headered.RoomVersion),
				},
			},
		})
		if err != nil {
			return "", fmt.Errorf("r.WriteOutputEvents: %w", err)
		}
	}
	return event.EventID(), nil
}
func ( r * Inputer ) processRoomEvent ( ctx context . Context , input * api . InputRoomEvent , ) ( eventID string , err error ) { headered := input . Event event := headered . Unwrap ( ) isRejected := false authEventNIDs , rejectionErr := helpers . CheckAuthEvents ( ctx , r . DB , headered , input . AuthEventIDs ) if rejectionErr != nil { logrus . WithError ( rejectionErr ) . WithField ( " " , event . EventID ( ) ) . WithField ( " " , input . AuthEventIDs ) . Error ( " " ) isRejected = true } var softfail bool if input . Kind == api . KindBackfill || input . Kind == api . KindNew { softfail , err = helpers . CheckForSoftFail ( ctx , r . DB , headered , input . StateEventIDs ) if err != nil { logrus . WithFields ( logrus . Fields { " " : event . EventID ( ) , " " : event . Type ( ) , " " : event . RoomID ( ) , } ) . WithError ( err ) . Info ( " " ) } } if input . TransactionID != nil { tdID := input . TransactionID eventID , err = r . DB . GetTransactionEventID ( ctx , tdID . TransactionID , tdID . SessionID , event . Sender ( ) , ) if err != nil || eventID != " " { return } } _ , stateAtEvent , redactionEvent , redactedEventID , err := r . DB . StoreEvent ( ctx , event , input . TransactionID , authEventNIDs , isRejected ) if err != nil { return " " , fmt . Errorf ( " " , err ) } if ! isRejected && redactedEventID == event . EventID ( ) { r , rerr := eventutil . RedactEvent ( redactionEvent , & event ) if rerr != nil { return " " , fmt . Errorf ( " " , rerr ) } event = * r } if input . Kind == api . KindOutlier { logrus . WithFields ( logrus . Fields { " " : event . EventID ( ) , " " : event . Type ( ) , " " : event . RoomID ( ) , " " : event . Sender ( ) , } ) . Debug ( " " ) return event . EventID ( ) , nil } roomInfo , err := r . DB . RoomInfo ( ctx , event . RoomID ( ) ) if err != nil { return " " , fmt . Errorf ( " " , err ) } if roomInfo == nil { return " " , fmt . Errorf ( " " , event . RoomID ( ) ) } if stateAtEvent . BeforeStateSnapshotNID == 0 { err = r . calculateAndSetState ( ctx , input , * roomInfo , & stateAtEvent , event , isRejected ) if err != nil { return " " , fmt . Errorf ( " " , err ) } } if isRejected || softfail { logrus . WithFields ( logrus . Fields { " " : event . EventID ( ) , " " : event . Type ( ) , " " : event . RoomID ( ) , " " : softfail , " " : event . Sender ( ) , } ) . Debug ( " " ) return event . EventID ( ) , rejectionErr } if input . Kind == api . KindRewrite { logrus . WithFields ( logrus . Fields { " " : event . EventID ( ) , " " : event . Type ( ) , " " : event . RoomID ( ) , } ) . Debug ( " " ) return event . EventID ( ) , nil } if err = r . updateLatestEvents ( ctx , roomInfo , stateAtEvent , event , input . SendAsServer , input . TransactionID , input . HasState , ) ; err != nil { return " " , fmt . Errorf ( " " , err ) } if redactedEventID != " " { err = r . WriteOutputEvents ( event . RoomID ( ) , [ ] api . OutputEvent { { Type : api . OutputTypeRedactedEvent , RedactedEvent : & api . OutputRedactedEvent { RedactedEventID : redactedEventID , RedactedBecause : redactionEvent . Headered ( headered . RoomVersion ) , } , } , } ) if err != nil { return " " , fmt . Errorf ( " " , err ) } } return event . EventID ( ) , nil }
processRoomEvent can only be called once at a time
processRoomEvent can only be called once at a time
// Parse and validate the event JSON // Check that the event passes authentication checks and work out // the numeric IDs for the auth events. // Check that the event passes authentication checks based on the // current room state. // If we don't have a transaction ID then get one. // On error OR event with the transaction already processed/processesing // Store the event. // if storing this event results in it being redacted then do so. // For outliers we can stop after we've stored the event itself as it // doesn't have any associated state to store and we don't need to // notify anyone about it. // We haven't calculated a state for this event yet. // Lets calculate one. // We stop here if the event is rejected: We've stored it but won't update forward extremities or notify anyone about it. // context // room info for the room being updated // state at event (below) // event // send as server // transaction ID // rewrites state? // processing this event resulted in an event (which may not be the one we're processing) // being redacted. We are guaranteed to have both sides (the redaction/redacted event), // so notify downstream components to redact this event - they should have it if they've // been tracking our output log. // Update the extremities of the event graph for the room