<commit_before>// Copyright 2017 Vector Creations Ltd
//
// 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 input contains the code processes new room events
package input

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"

	"github.com/Arceliar/phony"
	"github.com/getsentry/sentry-go"
	fedapi "github.com/matrix-org/dendrite/federationapi/api"
	"github.com/matrix-org/dendrite/roomserver/acls"
	"github.com/matrix-org/dendrite/roomserver/api"
	"github.com/matrix-org/dendrite/roomserver/internal/query"
	"github.com/matrix-org/dendrite/roomserver/storage"
	"github.com/matrix-org/dendrite/setup/jetstream"
	"github.com/matrix-org/gomatrixserverlib"
	"github.com/nats-io/nats.go"
	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
	"github.com/tidwall/gjson"
)

var keyContentFields = map[string]string{
	"m.room.join_rules":         "join_rule",
	"m.room.history_visibility": "history_visibility",
	"m.room.member":             "membership",
}

type Inputer struct {
	DB                   storage.Database
	JetStream            nats.JetStreamContext
	Durable              nats.SubOpt
	ServerName           gomatrixserverlib.ServerName
	FSAPI                fedapi.FederationInternalAPI
	KeyRing              gomatrixserverlib.JSONVerifier
	ACLs                 *acls.ServerACLs
	InputRoomEventTopic  string
	OutputRoomEventTopic string
	workers              sync.Map // room ID -> *phony.Inbox

	Queryer *query.Queryer
}

func (r *Inputer) workerForRoom(roomID string) *phony.Inbox {
	inbox, _ := r.workers.LoadOrStore(roomID, &phony.Inbox{})
	return inbox.(*phony.Inbox)
}

// eventsInProgress is an in-memory map to keep a track of which events we have
// queued up for processing. If we get a redelivery from NATS and we still have
// the queued up item then we won't do anything with the redelivered message. If
// we've restarted Dendrite and now this map is empty then it means that we will
// reload pending work from NATS.
var eventsInProgress sync.Map

// onMessage is called when a new event arrives in the roomserver input stream.
func (r *Inputer) Start() error {
	_, err := r.JetStream.Subscribe(
		r.InputRoomEventTopic,
		// We specifically don't use jetstream.WithJetStreamMessage here because we
		// queue the task off to a room-specific queue and the ACK needs to be sent
		// later, possibly with an error response to the inputter if synchronous.
		func(msg *nats.Msg) {
			roomID := msg.Header.Get("room_id")
			var inputRoomEvent api.InputRoomEvent
			if err := json.Unmarshal(msg.Data, &inputRoomEvent); err != nil {
				_ = msg.Term()
				return
			}

			_ = msg.InProgress()
			index := roomID + "\000" + inputRoomEvent.Event.EventID()
			if _, ok := eventsInProgress.LoadOrStore(index, struct{}{}); ok {
				// We're already waiting to deal with this event, so there's no
				// point in queuing it up again. We've notified NATS that we're
				// working on the message still, so that will have deferred the
				// redelivery by a bit.
				return
			}

			roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
			r.workerForRoom(roomID).Act(nil, func() {
				_ = msg.InProgress() // resets the acknowledgement wait timer
				defer eventsInProgress.Delete(index)
				defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
				if err := r.processRoomEvent(context.Background(), &inputRoomEvent); err != nil {
					if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
						sentry.CaptureException(err)
					}
				}
				_ = msg.Ack()
			})
		},
		// NATS wants to acknowledge automatically by default when the message is
		// read from the stream, but we want to override that behaviour by making
		// sure that we only acknowledge when we're happy we've done everything we
		// can. This ensures we retry things when it makes sense to do so.
		nats.ManualAck(),
		// Use a durable named consumer.
		r.Durable,
		// If we've missed things in the stream, e.g. we restarted, then replay
		// all of the queued messages that were waiting for us.
		nats.DeliverAll(),
		// Ensure that NATS doesn't try to resend us something that wasn't done
		// within the period of time that we might still be processing it.
		nats.AckWait((MaximumMissingProcessingTime*2)+(time.Second*10)),
	)
	return err
}

// InputRoomEvents implements api.RoomserverInternalAPI
func (r *Inputer) InputRoomEvents(
	ctx context.Context,
	request *api.InputRoomEventsRequest,
	response *api.InputRoomEventsResponse,
) {
	if request.Asynchronous {<commit_msg>Revert consumer change<commit_after>// Copyright 2017 Vector Creations Ltd
//
// 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 input contains the code processes new room events
package input

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"

	"github.com/Arceliar/phony"
	"github.com/getsentry/sentry-go"
	fedapi "github.com/matrix-org/dendrite/federationapi/api"
	"github.com/matrix-org/dendrite/roomserver/acls"
	"github.com/matrix-org/dendrite/roomserver/api"
	"github.com/matrix-org/dendrite/roomserver/internal/query"
	"github.com/matrix-org/dendrite/roomserver/storage"
	"github.com/matrix-org/dendrite/setup/jetstream"
	"github.com/matrix-org/gomatrixserverlib"
	"github.com/nats-io/nats.go"
	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
	"github.com/tidwall/gjson"
)

var keyContentFields = map[string]string{
	"m.room.join_rules":         "join_rule",
	"m.room.history_visibility": "history_visibility",
	"m.room.member":             "membership",
}

type Inputer struct {
	DB                   storage.Database
	JetStream            nats.JetStreamContext
	Durable              nats.SubOpt
	ServerName           gomatrixserverlib.ServerName
	FSAPI                fedapi.FederationInternalAPI
	KeyRing              gomatrixserverlib.JSONVerifier
	ACLs                 *acls.ServerACLs
	InputRoomEventTopic  string
	OutputRoomEventTopic string
	workers              sync.Map // room ID -> *phony.Inbox

	Queryer *query.Queryer
}

func (r *Inputer) workerForRoom(roomID string) *phony.Inbox {
	inbox, _ := r.workers.LoadOrStore(roomID, &phony.Inbox{})
	return inbox.(*phony.Inbox)
}

// eventsInProgress is an in-memory map to keep a track of which events we have
// queued up for processing. If we get a redelivery from NATS and we still have
// the queued up item then we won't do anything with the redelivered message. If
// we've restarted Dendrite and now this map is empty then it means that we will
// reload pending work from NATS.
var eventsInProgress sync.Map

// onMessage is called when a new event arrives in the roomserver input stream.
func (r *Inputer) Start() error {
	_, err := r.JetStream.Subscribe(
		r.InputRoomEventTopic,
		// We specifically don't use jetstream.WithJetStreamMessage here because we
		// queue the task off to a room-specific queue and the ACK needs to be sent
		// later, possibly with an error response to the inputter if synchronous.
		func(msg *nats.Msg) {
			roomID := msg.Header.Get("room_id")
			var inputRoomEvent api.InputRoomEvent
			if err := json.Unmarshal(msg.Data, &inputRoomEvent); err != nil {
				_ = msg.Term()
				return
			}

			_ = msg.InProgress()
			index := roomID + "\000" + inputRoomEvent.Event.EventID()
			if _, ok := eventsInProgress.LoadOrStore(index, struct{}{}); ok {
				// We're already waiting to deal with this event, so there's no
				// point in queuing it up again. We've notified NATS that we're
				// working on the message still, so that will have deferred the
				// redelivery by a bit.
				return
			}

			roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
			r.workerForRoom(roomID).Act(nil, func() {
				_ = msg.InProgress() // resets the acknowledgement wait timer
				defer eventsInProgress.Delete(index)
				defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
				if err := r.processRoomEvent(context.Background(), &inputRoomEvent); err != nil {
					if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
						sentry.CaptureException(err)
					}
				}
				_ = msg.Ack()
			})
		},
		// NATS wants to acknowledge automatically by default when the message is
		// read from the stream, but we want to override that behaviour by making
		// sure that we only acknowledge when we're happy we've done everything we
		// can. This ensures we retry things when it makes sense to do so.
		nats.ManualAck(),
		// Use a durable named consumer.
		r.Durable,
		// If we've missed things in the stream, e.g. we restarted, then replay
		// all of the queued messages that were waiting for us.
		nats.DeliverAll(),
		// Ensure that NATS doesn't try to resend us something that wasn't done
		// within the period of time that we might still be processing it.
		nats.AckWait(MaximumMissingProcessingTime+(time.Second*10)),
	)
	return err
}

// InputRoomEvents implements api.RoomserverInternalAPI
func (r *Inputer) InputRoomEvents(
	ctx context.Context,
	request *api.InputRoomEventsRequest,
	response *api.InputRoomEventsResponse,
) {
	if request.Asynchronous {
{'fraction_non_alphanumeric': 0.100166781124301, 'fraction_numerical': 0.002648876680074561, 'mean_word_length': 3.43410178338408, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3933223', 'n_tokens_mistral': 3372, 'n_tokens_neox': 3072, 'n_words': 996}