BREAKING: Make eduserver/appservice use userapi (#1138)

* BREAKING: Make eduserver/appservice use userapi

This is a breaking change because this PR restructures how the AS API
tracks its position in Kafka streams. Previously, it used the account DB
to store partition offsets. However, this is also being used by `clientapi`
for the same purpose, which is bad (each component needs to store offsets
independently or else you might lose messages across restarts). This PR
changes this behaviour to now store partition offsets in the `appservice`
database.

This means that:
 - Upon restart, the `appservice` component will attempt to replay all
   room events from the beginning of time.
 - An additional table will be created in the appservice database, which
   in and of itself is backwards compatible.

* Return ErrorConflict
This commit is contained in:
Kegsay 2020-06-16 17:39:56 +01:00 committed by GitHub
parent 83391da0e0
commit e15a8042a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 207 additions and 53 deletions

View File

@ -16,7 +16,6 @@ package appservice
import ( import (
"context" "context"
"errors"
"net/http" "net/http"
"sync" "sync"
"time" "time"
@ -29,12 +28,10 @@ import (
"github.com/matrix-org/dendrite/appservice/storage" "github.com/matrix-org/dendrite/appservice/storage"
"github.com/matrix-org/dendrite/appservice/types" "github.com/matrix-org/dendrite/appservice/types"
"github.com/matrix-org/dendrite/appservice/workers" "github.com/matrix-org/dendrite/appservice/workers"
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/setup" "github.com/matrix-org/dendrite/internal/setup"
"github.com/matrix-org/dendrite/internal/sqlutil"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
@ -47,8 +44,7 @@ func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceQuer
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes. // can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI( func NewInternalAPI(
base *setup.BaseDendrite, base *setup.BaseDendrite,
accountsDB accounts.Database, userAPI userapi.UserInternalAPI,
deviceDB devices.Database,
rsAPI roomserverAPI.RoomserverInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI,
) appserviceAPI.AppServiceQueryAPI { ) appserviceAPI.AppServiceQueryAPI {
// Create a connection to the appservice postgres DB // Create a connection to the appservice postgres DB
@ -70,7 +66,7 @@ func NewInternalAPI(
workerStates[i] = ws workerStates[i] = ws
// Create bot account for this AS if it doesn't already exist // Create bot account for this AS if it doesn't already exist
if err = generateAppServiceAccount(accountsDB, deviceDB, appservice); err != nil { if err = generateAppServiceAccount(userAPI, appservice); err != nil {
logrus.WithFields(logrus.Fields{ logrus.WithFields(logrus.Fields{
"appservice": appservice.ID, "appservice": appservice.ID,
}).WithError(err).Panicf("failed to generate bot account for appservice") }).WithError(err).Panicf("failed to generate bot account for appservice")
@ -90,7 +86,7 @@ func NewInternalAPI(
// We can't add ASes at runtime so this is safe to do. // We can't add ASes at runtime so this is safe to do.
if len(workerStates) > 0 { if len(workerStates) > 0 {
consumer := consumers.NewOutputRoomEventConsumer( consumer := consumers.NewOutputRoomEventConsumer(
base.Cfg, base.KafkaConsumer, accountsDB, appserviceDB, base.Cfg, base.KafkaConsumer, appserviceDB,
rsAPI, workerStates, rsAPI, workerStates,
) )
if err := consumer.Start(); err != nil { if err := consumer.Start(); err != nil {
@ -109,22 +105,24 @@ func NewInternalAPI(
// `sender_localpart` field of each application service if it doesn't // `sender_localpart` field of each application service if it doesn't
// exist already // exist already
func generateAppServiceAccount( func generateAppServiceAccount(
accountsDB accounts.Database, userAPI userapi.UserInternalAPI,
deviceDB devices.Database,
as config.ApplicationService, as config.ApplicationService,
) error { ) error {
ctx := context.Background() var accRes userapi.PerformAccountCreationResponse
err := userAPI.PerformAccountCreation(context.Background(), &userapi.PerformAccountCreationRequest{
// Create an account for the application service Localpart: as.SenderLocalpart,
_, err := accountsDB.CreateAccount(ctx, as.SenderLocalpart, "", as.ID) AppServiceID: as.ID,
OnConflict: userapi.ConflictUpdate,
}, &accRes)
if err != nil { if err != nil {
if errors.Is(err, sqlutil.ErrUserExists) { // This account already exists
return nil
}
return err return err
} }
var devRes userapi.PerformDeviceCreationResponse
// Create a dummy device with a dummy token for the application service err = userAPI.PerformDeviceCreation(context.Background(), &userapi.PerformDeviceCreationRequest{
_, err = deviceDB.CreateDevice(ctx, as.SenderLocalpart, nil, as.ASToken, &as.SenderLocalpart) Localpart: as.SenderLocalpart,
AccessToken: as.ASToken,
DeviceID: &as.SenderLocalpart,
DeviceDisplayName: &as.SenderLocalpart,
}, &devRes)
return err return err
} }

View File

@ -20,7 +20,6 @@ import (
"github.com/matrix-org/dendrite/appservice/storage" "github.com/matrix-org/dendrite/appservice/storage"
"github.com/matrix-org/dendrite/appservice/types" "github.com/matrix-org/dendrite/appservice/types"
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
"github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/api"
@ -33,7 +32,6 @@ import (
// OutputRoomEventConsumer consumes events that originated in the room server. // OutputRoomEventConsumer consumes events that originated in the room server.
type OutputRoomEventConsumer struct { type OutputRoomEventConsumer struct {
roomServerConsumer *internal.ContinualConsumer roomServerConsumer *internal.ContinualConsumer
db accounts.Database
asDB storage.Database asDB storage.Database
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
serverName string serverName string
@ -45,7 +43,6 @@ type OutputRoomEventConsumer struct {
func NewOutputRoomEventConsumer( func NewOutputRoomEventConsumer(
cfg *config.Dendrite, cfg *config.Dendrite,
kafkaConsumer sarama.Consumer, kafkaConsumer sarama.Consumer,
store accounts.Database,
appserviceDB storage.Database, appserviceDB storage.Database,
rsAPI api.RoomserverInternalAPI, rsAPI api.RoomserverInternalAPI,
workerStates []types.ApplicationServiceWorkerState, workerStates []types.ApplicationServiceWorkerState,
@ -53,11 +50,10 @@ func NewOutputRoomEventConsumer(
consumer := internal.ContinualConsumer{ consumer := internal.ContinualConsumer{
Topic: string(cfg.Kafka.Topics.OutputRoomEvent), Topic: string(cfg.Kafka.Topics.OutputRoomEvent),
Consumer: kafkaConsumer, Consumer: kafkaConsumer,
PartitionStore: store, PartitionStore: appserviceDB,
} }
s := &OutputRoomEventConsumer{ s := &OutputRoomEventConsumer{
roomServerConsumer: &consumer, roomServerConsumer: &consumer,
db: store,
asDB: appserviceDB, asDB: appserviceDB,
rsAPI: rsAPI, rsAPI: rsAPI,
serverName: string(cfg.Matrix.ServerName), serverName: string(cfg.Matrix.ServerName),

View File

@ -17,10 +17,12 @@ package storage
import ( import (
"context" "context"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
) )
type Database interface { type Database interface {
internal.PartitionStorer
StoreEvent(ctx context.Context, appServiceID string, event *gomatrixserverlib.HeaderedEvent) error StoreEvent(ctx context.Context, appServiceID string, event *gomatrixserverlib.HeaderedEvent) error
GetEventsWithAppServiceID(ctx context.Context, appServiceID string, limit int) (int, int, []gomatrixserverlib.HeaderedEvent, bool, error) GetEventsWithAppServiceID(ctx context.Context, appServiceID string, limit int) (int, int, []gomatrixserverlib.HeaderedEvent, bool, error)
CountEventsWithAppServiceID(ctx context.Context, appServiceID string) (int, error) CountEventsWithAppServiceID(ctx context.Context, appServiceID string) (int, error)

View File

@ -27,6 +27,7 @@ import (
// Database stores events intended to be later sent to application services // Database stores events intended to be later sent to application services
type Database struct { type Database struct {
sqlutil.PartitionOffsetStatements
events eventsStatements events eventsStatements
txnID txnStatements txnID txnStatements
db *sql.DB db *sql.DB
@ -42,6 +43,9 @@ func NewDatabase(dataSourceName string, dbProperties sqlutil.DbProperties) (*Dat
if err = result.prepare(); err != nil { if err = result.prepare(); err != nil {
return nil, err return nil, err
} }
if err = result.PartitionOffsetStatements.Prepare(result.db, "appservice"); err != nil {
return nil, err
}
return &result, nil return &result, nil
} }

View File

@ -27,6 +27,7 @@ import (
// Database stores events intended to be later sent to application services // Database stores events intended to be later sent to application services
type Database struct { type Database struct {
sqlutil.PartitionOffsetStatements
events eventsStatements events eventsStatements
txnID txnStatements txnID txnStatements
db *sql.DB db *sql.DB
@ -46,6 +47,9 @@ func NewDatabase(dataSourceName string) (*Database, error) {
if err = result.prepare(); err != nil { if err = result.prepare(); err != nil {
return nil, err return nil, err
} }
if err = result.PartitionOffsetStatements.Prepare(result.db, "appservice"); err != nil {
return nil, err
}
return &result, nil return &result, nil
} }

View File

@ -40,6 +40,10 @@ type Database interface {
GetMembershipsByLocalpart(ctx context.Context, localpart string) (memberships []authtypes.Membership, err error) GetMembershipsByLocalpart(ctx context.Context, localpart string) (memberships []authtypes.Membership, err error)
SaveAccountData(ctx context.Context, localpart, roomID, dataType, content string) error SaveAccountData(ctx context.Context, localpart, roomID, dataType, content string) error
GetAccountData(ctx context.Context, localpart string) (global []gomatrixserverlib.ClientEvent, rooms map[string][]gomatrixserverlib.ClientEvent, err error) GetAccountData(ctx context.Context, localpart string) (global []gomatrixserverlib.ClientEvent, rooms map[string][]gomatrixserverlib.ClientEvent, err error)
// GetAccountDataByType returns account data matching a given
// localpart, room ID and type.
// If no account data could be found, returns nil
// Returns an error if there was an issue with the retrieval
GetAccountDataByType(ctx context.Context, localpart, roomID, dataType string) (data *gomatrixserverlib.ClientEvent, err error) GetAccountDataByType(ctx context.Context, localpart, roomID, dataType string) (data *gomatrixserverlib.ClientEvent, err error)
GetNewNumericLocalpart(ctx context.Context) (int64, error) GetNewNumericLocalpart(ctx context.Context) (int64, error)
SaveThreePIDAssociation(ctx context.Context, threepid, localpart, medium string) (err error) SaveThreePIDAssociation(ctx context.Context, threepid, localpart, medium string) (err error)

View File

@ -24,6 +24,12 @@ type Database interface {
GetDeviceByAccessToken(ctx context.Context, token string) (*api.Device, error) GetDeviceByAccessToken(ctx context.Context, token string) (*api.Device, error)
GetDeviceByID(ctx context.Context, localpart, deviceID string) (*api.Device, error) GetDeviceByID(ctx context.Context, localpart, deviceID string) (*api.Device, error)
GetDevicesByLocalpart(ctx context.Context, localpart string) ([]api.Device, error) GetDevicesByLocalpart(ctx context.Context, localpart string) ([]api.Device, error)
// CreateDevice makes a new device associated with the given user ID localpart.
// If there is already a device with the same device ID for this user, that access token will be revoked
// and replaced with the given accessToken. If the given accessToken is already in use for another device,
// an error will be returned.
// If no device ID is given one is generated.
// Returns the device on success.
CreateDevice(ctx context.Context, localpart string, deviceID *string, accessToken string, displayName *string) (dev *api.Device, returnErr error) CreateDevice(ctx context.Context, localpart string, deviceID *string, accessToken string, displayName *string) (dev *api.Device, returnErr error)
UpdateDevice(ctx context.Context, localpart, deviceID string, displayName *string) error UpdateDevice(ctx context.Context, localpart, deviceID string, displayName *string) error
RemoveDevice(ctx context.Context, deviceID, localpart string) error RemoveDevice(ctx context.Context, deviceID, localpart string) error

View File

@ -24,11 +24,10 @@ func main() {
base := setup.NewBaseDendrite(cfg, "AppServiceAPI", true) base := setup.NewBaseDendrite(cfg, "AppServiceAPI", true)
defer base.Close() // nolint: errcheck defer base.Close() // nolint: errcheck
accountDB := base.CreateAccountsDB() userAPI := base.UserAPIClient()
deviceDB := base.CreateDeviceDB()
rsAPI := base.RoomserverHTTPClient() rsAPI := base.RoomserverHTTPClient()
intAPI := appservice.NewInternalAPI(base, accountDB, deviceDB, rsAPI) intAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
appservice.AddInternalRoutes(base.InternalAPIMux, intAPI) appservice.AddInternalRoutes(base.InternalAPIMux, intAPI)
base.SetupAndServeHTTP(string(base.Cfg.Bind.AppServiceAPI), string(base.Cfg.Listen.AppServiceAPI)) base.SetupAndServeHTTP(string(base.Cfg.Bind.AppServiceAPI), string(base.Cfg.Listen.AppServiceAPI))

View File

@ -141,6 +141,7 @@ func main() {
accountDB := base.Base.CreateAccountsDB() accountDB := base.Base.CreateAccountsDB()
deviceDB := base.Base.CreateDeviceDB() deviceDB := base.Base.CreateDeviceDB()
federation := createFederationClient(base) federation := createFederationClient(base)
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
serverKeyAPI := serverkeyapi.NewInternalAPI( serverKeyAPI := serverkeyapi.NewInternalAPI(
base.Base.Cfg, federation, base.Base.Caches, base.Base.Cfg, federation, base.Base.Caches,
@ -154,9 +155,9 @@ func main() {
&base.Base, keyRing, federation, &base.Base, keyRing, federation,
) )
eduInputAPI := eduserver.NewInternalAPI( eduInputAPI := eduserver.NewInternalAPI(
&base.Base, cache.New(), deviceDB, &base.Base, cache.New(), userAPI,
) )
asAPI := appservice.NewInternalAPI(&base.Base, accountDB, deviceDB, rsAPI) asAPI := appservice.NewInternalAPI(&base.Base, userAPI, rsAPI)
fsAPI := federationsender.NewInternalAPI( fsAPI := federationsender.NewInternalAPI(
&base.Base, federation, rsAPI, keyRing, &base.Base, federation, rsAPI, keyRing,
) )
@ -165,7 +166,6 @@ func main() {
if err != nil { if err != nil {
logrus.WithError(err).Panicf("failed to connect to public rooms db") logrus.WithError(err).Panicf("failed to connect to public rooms db")
} }
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
monolith := setup.Monolith{ monolith := setup.Monolith{
Config: base.Base.Cfg, Config: base.Base.Cfg,

View File

@ -130,16 +130,18 @@ func main() {
serverKeyAPI := &signing.YggdrasilKeys{} serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing() keyRing := serverKeyAPI.KeyRing()
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
rsComponent := roomserver.NewInternalAPI( rsComponent := roomserver.NewInternalAPI(
base, keyRing, federation, base, keyRing, federation,
) )
rsAPI := rsComponent rsAPI := rsComponent
eduInputAPI := eduserver.NewInternalAPI( eduInputAPI := eduserver.NewInternalAPI(
base, cache.New(), deviceDB, base, cache.New(), userAPI,
) )
asAPI := appservice.NewInternalAPI(base, accountDB, deviceDB, rsAPI) asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
fsAPI := federationsender.NewInternalAPI( fsAPI := federationsender.NewInternalAPI(
base, federation, rsAPI, keyRing, base, federation, rsAPI, keyRing,
@ -153,7 +155,6 @@ func main() {
} }
embed.Embed(*instancePort, "Yggdrasil Demo") embed.Embed(*instancePort, "Yggdrasil Demo")
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
monolith := setup.Monolith{ monolith := setup.Monolith{
Config: base.Cfg, Config: base.Cfg,

View File

@ -29,9 +29,8 @@ func main() {
logrus.WithError(err).Warn("BaseDendrite close failed") logrus.WithError(err).Warn("BaseDendrite close failed")
} }
}() }()
deviceDB := base.CreateDeviceDB()
intAPI := eduserver.NewInternalAPI(base, cache.New(), deviceDB) intAPI := eduserver.NewInternalAPI(base, cache.New(), base.UserAPIClient())
eduserver.AddInternalRoutes(base.InternalAPIMux, intAPI) eduserver.AddInternalRoutes(base.InternalAPIMux, intAPI)
base.SetupAndServeHTTP(string(base.Cfg.Bind.EDUServer), string(base.Cfg.Listen.EDUServer)) base.SetupAndServeHTTP(string(base.Cfg.Bind.EDUServer), string(base.Cfg.Listen.EDUServer))

View File

@ -75,6 +75,7 @@ func main() {
serverKeyAPI = base.ServerKeyAPIClient() serverKeyAPI = base.ServerKeyAPIClient()
} }
keyRing := serverKeyAPI.KeyRing() keyRing := serverKeyAPI.KeyRing()
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, cfg.Derived.ApplicationServices)
rsImpl := roomserver.NewInternalAPI( rsImpl := roomserver.NewInternalAPI(
base, keyRing, federation, base, keyRing, federation,
@ -92,14 +93,14 @@ func main() {
} }
eduInputAPI := eduserver.NewInternalAPI( eduInputAPI := eduserver.NewInternalAPI(
base, cache.New(), deviceDB, base, cache.New(), userAPI,
) )
if base.UseHTTPAPIs { if base.UseHTTPAPIs {
eduserver.AddInternalRoutes(base.InternalAPIMux, eduInputAPI) eduserver.AddInternalRoutes(base.InternalAPIMux, eduInputAPI)
eduInputAPI = base.EDUServerClient() eduInputAPI = base.EDUServerClient()
} }
asAPI := appservice.NewInternalAPI(base, accountDB, deviceDB, rsAPI) asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
if base.UseHTTPAPIs { if base.UseHTTPAPIs {
appservice.AddInternalRoutes(base.InternalAPIMux, asAPI) appservice.AddInternalRoutes(base.InternalAPIMux, asAPI)
asAPI = base.AppserviceHTTPClient() asAPI = base.AppserviceHTTPClient()
@ -121,8 +122,6 @@ func main() {
logrus.WithError(err).Panicf("failed to connect to public rooms db") logrus.WithError(err).Panicf("failed to connect to public rooms db")
} }
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, cfg.Derived.ApplicationServices)
monolith := setup.Monolith{ monolith := setup.Monolith{
Config: base.Cfg, Config: base.Cfg,
AccountDB: accountDB, AccountDB: accountDB,

View File

@ -194,6 +194,7 @@ func main() {
accountDB := base.CreateAccountsDB() accountDB := base.CreateAccountsDB()
deviceDB := base.CreateDeviceDB() deviceDB := base.CreateDeviceDB()
federation := createFederationClient(cfg, node) federation := createFederationClient(cfg, node)
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
fetcher := &libp2pKeyFetcher{} fetcher := &libp2pKeyFetcher{}
keyRing := gomatrixserverlib.KeyRing{ keyRing := gomatrixserverlib.KeyRing{
@ -204,9 +205,9 @@ func main() {
} }
rsAPI := roomserver.NewInternalAPI(base, keyRing, federation) rsAPI := roomserver.NewInternalAPI(base, keyRing, federation)
eduInputAPI := eduserver.NewInternalAPI(base, cache.New(), deviceDB) eduInputAPI := eduserver.NewInternalAPI(base, cache.New(), userAPI)
asQuery := appservice.NewInternalAPI( asQuery := appservice.NewInternalAPI(
base, accountDB, deviceDB, rsAPI, base, userAPI, rsAPI,
) )
fedSenderAPI := federationsender.NewInternalAPI(base, federation, rsAPI, &keyRing) fedSenderAPI := federationsender.NewInternalAPI(base, federation, rsAPI, &keyRing)
rsAPI.SetFederationSenderAPI(fedSenderAPI) rsAPI.SetFederationSenderAPI(fedSenderAPI)
@ -217,8 +218,6 @@ func main() {
logrus.WithError(err).Panicf("failed to connect to public rooms db") logrus.WithError(err).Panicf("failed to connect to public rooms db")
} }
userAPI := userapi.NewInternalAPI(accountDB, deviceDB, cfg.Matrix.ServerName, nil)
monolith := setup.Monolith{ monolith := setup.Monolith{
Config: base.Cfg, Config: base.Cfg,
AccountDB: accountDB, AccountDB: accountDB,

View File

@ -18,12 +18,12 @@ package eduserver
import ( import (
"github.com/gorilla/mux" "github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/eduserver/cache" "github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/eduserver/input" "github.com/matrix-org/dendrite/eduserver/input"
"github.com/matrix-org/dendrite/eduserver/inthttp" "github.com/matrix-org/dendrite/eduserver/inthttp"
"github.com/matrix-org/dendrite/internal/setup" "github.com/matrix-org/dendrite/internal/setup"
userapi "github.com/matrix-org/dendrite/userapi/api"
) )
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions // AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@ -37,11 +37,11 @@ func AddInternalRoutes(internalMux *mux.Router, inputAPI api.EDUServerInputAPI)
func NewInternalAPI( func NewInternalAPI(
base *setup.BaseDendrite, base *setup.BaseDendrite,
eduCache *cache.EDUCache, eduCache *cache.EDUCache,
deviceDB devices.Database, userAPI userapi.UserInternalAPI,
) api.EDUServerInputAPI { ) api.EDUServerInputAPI {
return &input.EDUServerInputAPI{ return &input.EDUServerInputAPI{
Cache: eduCache, Cache: eduCache,
DeviceDB: deviceDB, UserAPI: userAPI,
Producer: base.KafkaProducer, Producer: base.KafkaProducer,
OutputTypingEventTopic: string(base.Cfg.Kafka.Topics.OutputTypingEvent), OutputTypingEventTopic: string(base.Cfg.Kafka.Topics.OutputTypingEvent),
OutputSendToDeviceEventTopic: string(base.Cfg.Kafka.Topics.OutputSendToDeviceEvent), OutputSendToDeviceEventTopic: string(base.Cfg.Kafka.Topics.OutputSendToDeviceEvent),

View File

@ -22,9 +22,9 @@ import (
"time" "time"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/eduserver/cache" "github.com/matrix-org/dendrite/eduserver/cache"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
@ -39,8 +39,8 @@ type EDUServerInputAPI struct {
OutputSendToDeviceEventTopic string OutputSendToDeviceEventTopic string
// kafka producer // kafka producer
Producer sarama.SyncProducer Producer sarama.SyncProducer
// device database // Internal user query API
DeviceDB devices.Database UserAPI userapi.UserInternalAPI
// our server name // our server name
ServerName gomatrixserverlib.ServerName ServerName gomatrixserverlib.ServerName
} }
@ -115,7 +115,7 @@ func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error {
func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) error { func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) error {
devices := []string{} devices := []string{}
localpart, domain, err := gomatrixserverlib.SplitID('@', ise.UserID) _, domain, err := gomatrixserverlib.SplitID('@', ise.UserID)
if err != nil { if err != nil {
return err return err
} }
@ -126,11 +126,14 @@ func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) e
// wildcard as we don't know about the remote devices, so instead we leave it // wildcard as we don't know about the remote devices, so instead we leave it
// as-is, so that the federation sender can send it on with the wildcard intact. // as-is, so that the federation sender can send it on with the wildcard intact.
if domain == t.ServerName && ise.DeviceID == "*" { if domain == t.ServerName && ise.DeviceID == "*" {
devs, err := t.DeviceDB.GetDevicesByLocalpart(context.TODO(), localpart) var res userapi.QueryDevicesResponse
err = t.UserAPI.QueryDevices(context.TODO(), &userapi.QueryDevicesRequest{
UserID: ise.UserID,
}, &res)
if err != nil { if err != nil {
return err return err
} }
for _, dev := range devs { for _, dev := range res.Devices {
devices = append(devices, dev.ID) devices = append(devices, dev.ID)
} }
} else { } else {

View File

@ -22,6 +22,8 @@ import (
// UserInternalAPI is the internal API for information about users and devices. // UserInternalAPI is the internal API for information about users and devices.
type UserInternalAPI interface { type UserInternalAPI interface {
PerformAccountCreation(ctx context.Context, req *PerformAccountCreationRequest, res *PerformAccountCreationResponse) error
PerformDeviceCreation(ctx context.Context, req *PerformDeviceCreationRequest, res *PerformDeviceCreationResponse) error
QueryProfile(ctx context.Context, req *QueryProfileRequest, res *QueryProfileResponse) error QueryProfile(ctx context.Context, req *QueryProfileRequest, res *QueryProfileResponse) error
QueryAccessToken(ctx context.Context, req *QueryAccessTokenRequest, res *QueryAccessTokenResponse) error QueryAccessToken(ctx context.Context, req *QueryAccessTokenRequest, res *QueryAccessTokenResponse) error
QueryDevices(ctx context.Context, req *QueryDevicesRequest, res *QueryDevicesResponse) error QueryDevices(ctx context.Context, req *QueryDevicesRequest, res *QueryDevicesResponse) error
@ -85,6 +87,38 @@ type QueryProfileResponse struct {
AvatarURL string AvatarURL string
} }
// PerformAccountCreationRequest is the request for PerformAccountCreation
type PerformAccountCreationRequest struct {
Localpart string
AppServiceID string
Password string
OnConflict Conflict
}
// PerformAccountCreationResponse is the response for PerformAccountCreation
type PerformAccountCreationResponse struct {
AccountCreated bool
UserID string
}
// PerformDeviceCreationRequest is the request for PerformDeviceCreation
type PerformDeviceCreationRequest struct {
Localpart string
AccessToken string // optional: if blank one will be made on your behalf
// optional: if nil an ID is generated for you. If set, replaces any existing device session,
// which will generate a new access token and invalidate the old one.
DeviceID *string
// optional: if nil no display name will be associated with this device.
DeviceDisplayName *string
}
// PerformDeviceCreationResponse is the response for PerformDeviceCreation
type PerformDeviceCreationResponse struct {
DeviceCreated bool
AccessToken string
DeviceID string
}
// Device represents a client's device (mobile, web, etc) // Device represents a client's device (mobile, web, etc)
type Device struct { type Device struct {
ID string ID string
@ -108,3 +142,22 @@ type ErrorForbidden struct {
func (e *ErrorForbidden) Error() string { func (e *ErrorForbidden) Error() string {
return "Forbidden: " + e.Message return "Forbidden: " + e.Message
} }
// ErrorConflict is an error indicating that there was a conflict which resulted in the request being aborted.
type ErrorConflict struct {
Message string
}
func (e *ErrorConflict) Error() string {
return "Conflict: " + e.Message
}
// Conflict is an enum representing what to do when encountering conflicting when creating profiles/devices
type Conflict int
const (
// ConflictUpdate will update matching records returning no error
ConflictUpdate Conflict = 1
// ConflictAbort will reject the request with ErrorConflict
ConflictAbort Conflict = 2
)

View File

@ -17,6 +17,7 @@ package internal
import ( import (
"context" "context"
"database/sql" "database/sql"
"errors"
"fmt" "fmt"
"github.com/matrix-org/dendrite/appservice/types" "github.com/matrix-org/dendrite/appservice/types"
@ -24,6 +25,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices" "github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/clientapi/userutil" "github.com/matrix-org/dendrite/clientapi/userutil"
"github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
) )
@ -36,6 +38,38 @@ type UserInternalAPI struct {
AppServices []config.ApplicationService AppServices []config.ApplicationService
} }
func (a *UserInternalAPI) PerformAccountCreation(ctx context.Context, req *api.PerformAccountCreationRequest, res *api.PerformAccountCreationResponse) error {
acc, err := a.AccountDB.CreateAccount(ctx, req.Localpart, req.Password, req.AppServiceID)
if err != nil {
if errors.Is(err, sqlutil.ErrUserExists) { // This account already exists
switch req.OnConflict {
case api.ConflictUpdate:
break
case api.ConflictAbort:
return &api.ErrorConflict{
Message: err.Error(),
}
}
}
res.AccountCreated = false
res.UserID = fmt.Sprintf("@%s:%s", req.Localpart, a.ServerName)
return nil
}
res.AccountCreated = true
res.UserID = acc.UserID
return nil
}
func (a *UserInternalAPI) PerformDeviceCreation(ctx context.Context, req *api.PerformDeviceCreationRequest, res *api.PerformDeviceCreationResponse) error {
dev, err := a.DeviceDB.CreateDevice(ctx, req.Localpart, req.DeviceID, req.AccessToken, req.DeviceDisplayName)
if err != nil {
return err
}
res.DeviceCreated = true
res.AccessToken = dev.AccessToken
res.DeviceID = dev.ID
return nil
}
func (a *UserInternalAPI) QueryProfile(ctx context.Context, req *api.QueryProfileRequest, res *api.QueryProfileResponse) error { func (a *UserInternalAPI) QueryProfile(ctx context.Context, req *api.QueryProfileRequest, res *api.QueryProfileResponse) error {
local, domain, err := gomatrixserverlib.SplitID('@', req.UserID) local, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil { if err != nil {

View File

@ -26,6 +26,9 @@ import (
// HTTP paths for the internal HTTP APIs // HTTP paths for the internal HTTP APIs
const ( const (
PerformDeviceCreationPath = "/userapi/performDeviceCreation"
PerformAccountCreationPath = "/userapi/performAccountCreation"
QueryProfilePath = "/userapi/queryProfile" QueryProfilePath = "/userapi/queryProfile"
QueryAccessTokenPath = "/userapi/queryAccessToken" QueryAccessTokenPath = "/userapi/queryAccessToken"
QueryDevicesPath = "/userapi/queryDevices" QueryDevicesPath = "/userapi/queryDevices"
@ -52,6 +55,30 @@ type httpUserInternalAPI struct {
httpClient *http.Client httpClient *http.Client
} }
func (h *httpUserInternalAPI) PerformAccountCreation(
ctx context.Context,
request *api.PerformAccountCreationRequest,
response *api.PerformAccountCreationResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformAccountCreation")
defer span.Finish()
apiURL := h.apiURL + PerformAccountCreationPath
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
func (h *httpUserInternalAPI) PerformDeviceCreation(
ctx context.Context,
request *api.PerformDeviceCreationRequest,
response *api.PerformDeviceCreationResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformDeviceCreation")
defer span.Finish()
apiURL := h.apiURL + PerformDeviceCreationPath
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
func (h *httpUserInternalAPI) QueryProfile( func (h *httpUserInternalAPI) QueryProfile(
ctx context.Context, ctx context.Context,
request *api.QueryProfileRequest, request *api.QueryProfileRequest,

View File

@ -25,6 +25,32 @@ import (
) )
func AddRoutes(internalAPIMux *mux.Router, s api.UserInternalAPI) { func AddRoutes(internalAPIMux *mux.Router, s api.UserInternalAPI) {
internalAPIMux.Handle(PerformAccountCreationPath,
httputil.MakeInternalAPI("performAccountCreation", func(req *http.Request) util.JSONResponse {
request := api.PerformAccountCreationRequest{}
response := api.PerformAccountCreationResponse{}
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())
}
if err := s.PerformAccountCreation(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(PerformDeviceCreationPath,
httputil.MakeInternalAPI("performDeviceCreation", func(req *http.Request) util.JSONResponse {
request := api.PerformDeviceCreationRequest{}
response := api.PerformDeviceCreationResponse{}
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())
}
if err := s.PerformDeviceCreation(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(QueryProfilePath, internalAPIMux.Handle(QueryProfilePath,
httputil.MakeInternalAPI("queryProfile", func(req *http.Request) util.JSONResponse { httputil.MakeInternalAPI("queryProfile", func(req *http.Request) util.JSONResponse {
request := api.QueryProfileRequest{} request := api.QueryProfileRequest{}