1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
// Copyright 2024 New Vector Ltd.
// Copyright 2017 Vector Creations Ltd
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
package syncapi
import (
"context"
"github.com/element-hq/dendrite/internal/fulltext"
"github.com/element-hq/dendrite/internal/httputil"
"github.com/element-hq/dendrite/internal/sqlutil"
"github.com/element-hq/dendrite/setup/config"
"github.com/element-hq/dendrite/setup/process"
"github.com/sirupsen/logrus"
"github.com/element-hq/dendrite/internal/caching"
"github.com/element-hq/dendrite/roomserver/api"
"github.com/element-hq/dendrite/setup/jetstream"
userapi "github.com/element-hq/dendrite/userapi/api"
"github.com/element-hq/dendrite/syncapi/consumers"
"github.com/element-hq/dendrite/syncapi/notifier"
"github.com/element-hq/dendrite/syncapi/producers"
"github.com/element-hq/dendrite/syncapi/routing"
"github.com/element-hq/dendrite/syncapi/storage"
"github.com/element-hq/dendrite/syncapi/streams"
"github.com/element-hq/dendrite/syncapi/sync"
)
// AddPublicRoutes sets up and registers HTTP handlers for the SyncAPI
// component.
func AddPublicRoutes(
processContext *process.ProcessContext,
routers httputil.Routers,
dendriteCfg *config.Dendrite,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
userAPI userapi.SyncUserAPI,
rsAPI api.SyncRoomserverAPI,
caches caching.LazyLoadCache,
enableMetrics bool,
) {
js, natsClient := natsInstance.Prepare(processContext, &dendriteCfg.Global.JetStream)
syncDB, err := storage.NewSyncServerDatasource(processContext.Context(), cm, &dendriteCfg.SyncAPI.Database)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to sync db")
}
eduCache := caching.NewTypingCache()
notifier := notifier.NewNotifier(rsAPI)
streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, eduCache, caches, notifier)
notifier.SetCurrentPosition(streams.Latest(context.Background()))
if err = notifier.Load(context.Background(), syncDB); err != nil {
logrus.WithError(err).Panicf("failed to load notifier ")
}
var fts *fulltext.Search
if dendriteCfg.SyncAPI.Fulltext.Enabled {
fts, err = fulltext.New(processContext, dendriteCfg.SyncAPI.Fulltext)
if err != nil {
logrus.WithError(err).Panicf("failed to create full text")
}
}
federationPresenceProducer := &producers.FederationAPIPresenceProducer{
Topic: dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
JetStream: js,
}
presenceConsumer := consumers.NewPresenceConsumer(
processContext, &dendriteCfg.SyncAPI, js, natsClient, syncDB,
notifier, streams.PresenceStreamProvider,
userAPI,
)
requestPool := sync.NewRequestPool(syncDB, &dendriteCfg.SyncAPI, userAPI, rsAPI, streams, notifier, federationPresenceProducer, presenceConsumer, enableMetrics)
if err = presenceConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start presence consumer")
}
keyChangeConsumer := consumers.NewOutputKeyChangeEventConsumer(
processContext, &dendriteCfg.SyncAPI, dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
js, rsAPI, syncDB, notifier,
streams.DeviceListStreamProvider,
)
if err = keyChangeConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start key change consumer")
}
var asProducer *producers.AppserviceEventProducer
if len(dendriteCfg.AppServiceAPI.Derived.ApplicationServices) > 0 {
asProducer = &producers.AppserviceEventProducer{
JetStream: js, Topic: dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputAppserviceEvent),
}
}
roomConsumer := consumers.NewOutputRoomEventConsumer(
processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.PDUStreamProvider,
streams.InviteStreamProvider, rsAPI, fts, asProducer,
)
if err = roomConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start room server consumer")
}
clientConsumer := consumers.NewOutputClientDataConsumer(
processContext, &dendriteCfg.SyncAPI, js, natsClient, syncDB, notifier,
streams.AccountDataStreamProvider, fts,
)
if err = clientConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start client data consumer")
}
notificationConsumer := consumers.NewOutputNotificationDataConsumer(
processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.NotificationDataStreamProvider,
)
if err = notificationConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start notification data consumer")
}
typingConsumer := consumers.NewOutputTypingEventConsumer(
processContext, &dendriteCfg.SyncAPI, js, eduCache, notifier, streams.TypingStreamProvider,
)
if err = typingConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start typing consumer")
}
sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer(
processContext, &dendriteCfg.SyncAPI, js, syncDB, userAPI, notifier, streams.SendToDeviceStreamProvider,
)
if err = sendToDeviceConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start send-to-device consumer")
}
receiptConsumer := consumers.NewOutputReceiptEventConsumer(
processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.ReceiptStreamProvider,
)
if err = receiptConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start receipts consumer")
}
rateLimits := httputil.NewRateLimits(&dendriteCfg.ClientAPI.RateLimiting)
routing.Setup(
routers.Client, requestPool, syncDB, userAPI,
rsAPI, &dendriteCfg.SyncAPI, caches, fts,
rateLimits,
)
}
|
