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

2 files changed, 10 insertions, 0 deletions

diff --git a/clientapi/auth/storage/accounts/interface.go b/clientapi/auth/storage/accounts/interface.go
index 4d1941a2..3391ccbf 100644
--- a/clientapi/auth/storage/accounts/interface.go
+++ b/clientapi/auth/storage/accounts/interface.go
@@ -40,6 +40,10 @@ type Database interface {
 	GetMembershipsByLocalpart(ctx context.Context, localpart string) (memberships []authtypes.Membership, err 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)
+	// 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)
 	GetNewNumericLocalpart(ctx context.Context) (int64, error)
 	SaveThreePIDAssociation(ctx context.Context, threepid, localpart, medium string) (err error)
diff --git a/clientapi/auth/storage/devices/interface.go b/clientapi/auth/storage/devices/interface.go
index fc2f4a32..4bdb5785 100644
--- a/clientapi/auth/storage/devices/interface.go
+++ b/clientapi/auth/storage/devices/interface.go
@@ -24,6 +24,12 @@ type Database interface {
 	GetDeviceByAccessToken(ctx context.Context, token string) (*api.Device, error)
 	GetDeviceByID(ctx context.Context, localpart, deviceID 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)
 	UpdateDevice(ctx context.Context, localpart, deviceID string, displayName *string) error
 	RemoveDevice(ctx context.Context, deviceID, localpart string) error