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

4 files changed, 140 insertions, 0 deletions

diff --git a/userapi/api/api.go b/userapi/api/api.go
index 1578268a..34c74bb3 100644
--- a/userapi/api/api.go
+++ b/userapi/api/api.go
@@ -22,6 +22,8 @@ import (
 
 // UserInternalAPI is the internal API for information about users and devices.
 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
 	QueryAccessToken(ctx context.Context, req *QueryAccessTokenRequest, res *QueryAccessTokenResponse) error
 	QueryDevices(ctx context.Context, req *QueryDevicesRequest, res *QueryDevicesResponse) error
@@ -85,6 +87,38 @@ type QueryProfileResponse struct {
 	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)
 type Device struct {
 	ID     string
@@ -108,3 +142,22 @@ type ErrorForbidden struct {
 func (e *ErrorForbidden) Error() string {
 	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
+)
diff --git a/userapi/internal/api.go b/userapi/internal/api.go
index 6e737b81..1b34dc7b 100644
--- a/userapi/internal/api.go
+++ b/userapi/internal/api.go
@@ -17,6 +17,7 @@ package internal
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 
 	"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/userutil"
 	"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/gomatrixserverlib"
 )
@@ -36,6 +38,38 @@ type UserInternalAPI struct {
 	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 {
 	local, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
 	if err != nil {
diff --git a/userapi/inthttp/client.go b/userapi/inthttp/client.go
index 48e6d7d7..0e9628c5 100644
--- a/userapi/inthttp/client.go
+++ b/userapi/inthttp/client.go
@@ -26,6 +26,9 @@ import (
 
 // HTTP paths for the internal HTTP APIs
 const (
+	PerformDeviceCreationPath  = "/userapi/performDeviceCreation"
+	PerformAccountCreationPath = "/userapi/performAccountCreation"
+
 	QueryProfilePath     = "/userapi/queryProfile"
 	QueryAccessTokenPath = "/userapi/queryAccessToken"
 	QueryDevicesPath     = "/userapi/queryDevices"
@@ -52,6 +55,30 @@ type httpUserInternalAPI struct {
 	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(
 	ctx context.Context,
 	request *api.QueryProfileRequest,
diff --git a/userapi/inthttp/server.go b/userapi/inthttp/server.go
index 8bf2efc0..8f3be773 100644
--- a/userapi/inthttp/server.go
+++ b/userapi/inthttp/server.go
@@ -25,6 +25,32 @@ import (
 )
 
 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,
 		httputil.MakeInternalAPI("queryProfile", func(req *http.Request) util.JSONResponse {
 			request := api.QueryProfileRequest{}