Implement interfaces for syncapi storage

syncapi/consumers/clientapi.go

@@ -30,7 +30,7 @@ import (
 // OutputClientDataConsumer consumes events that originated in the client API server.
 type OutputClientDataConsumer struct {
 	clientAPIConsumer *common.ContinualConsumer
-	db                *storage.SyncServerDatasource
+	db                storage.Database
 	notifier          *sync.Notifier
 }
 
@@ -39,7 +39,7 @@ func NewOutputClientDataConsumer(
 	cfg *config.Dendrite,
 	kafkaConsumer sarama.Consumer,
 	n *sync.Notifier,
-	store *storage.SyncServerDatasource,
+	store storage.Database,
 ) *OutputClientDataConsumer {
 
 	consumer := common.ContinualConsumer{

syncapi/consumers/roomserver.go

@@ -33,7 +33,7 @@ import (
 // OutputRoomEventConsumer consumes events that originated in the room server.
 type OutputRoomEventConsumer struct {
 	roomServerConsumer *common.ContinualConsumer
-	db                 *storage.SyncServerDatasource
+	db                 storage.Database
 	notifier           *sync.Notifier
 	query              api.RoomserverQueryAPI
 }
@@ -43,7 +43,7 @@ func NewOutputRoomEventConsumer(
 	cfg *config.Dendrite,
 	kafkaConsumer sarama.Consumer,
 	n *sync.Notifier,
-	store *storage.SyncServerDatasource,
+	store storage.Database,
 	queryAPI api.RoomserverQueryAPI,
 ) *OutputRoomEventConsumer {
 

syncapi/consumers/typingserver.go

@@ -30,7 +30,7 @@ import (
 // OutputTypingEventConsumer consumes events that originated in the typing server.
 type OutputTypingEventConsumer struct {
 	typingConsumer *common.ContinualConsumer
-	db             *storage.SyncServerDatasource
+	db             storage.Database
 	notifier       *sync.Notifier
 }
 
@@ -40,7 +40,7 @@ func NewOutputTypingEventConsumer(
 	cfg *config.Dendrite,
 	kafkaConsumer sarama.Consumer,
 	n *sync.Notifier,
-	store *storage.SyncServerDatasource,
+	store storage.Database,
 ) *OutputTypingEventConsumer {
 
 	consumer := common.ContinualConsumer{

syncapi/routing/routing.go

@@ -34,7 +34,7 @@ const pathPrefixR0 = "/_matrix/client/r0"
 // Due to Setup being used to call many other functions, a gocyclo nolint is
 // applied:
 // nolint: gocyclo
-func Setup(apiMux *mux.Router, srp *sync.RequestPool, syncDB *storage.SyncServerDatasource, deviceDB *devices.Database) {
+func Setup(apiMux *mux.Router, srp *sync.RequestPool, syncDB storage.Database, deviceDB *devices.Database) {
 	r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
 
 	authData := auth.Data{

syncapi/routing/state.go

@@ -40,7 +40,7 @@ type stateEventInStateResp struct {
 // TODO: Check if the user is in the room. If not, check if the room's history
 // is publicly visible. Current behaviour is returning an empty array if the
 // user cannot see the room's history.
-func OnIncomingStateRequest(req *http.Request, db *storage.SyncServerDatasource, roomID string) util.JSONResponse {
+func OnIncomingStateRequest(req *http.Request, db storage.Database, roomID string) util.JSONResponse {
 	// TODO(#287): Auth request and handle the case where the user has left (where
 	// we should return the state at the poin they left)
 
@@ -87,7 +87,7 @@ func OnIncomingStateRequest(req *http.Request, db *storage.SyncServerDatasource,
 // /rooms/{roomID}/state/{type}/{statekey} request. It will look in current
 // state to see if there is an event with that type and state key, if there
 // is then (by default) we return the content, otherwise a 404.
-func OnIncomingStateTypeRequest(req *http.Request, db *storage.SyncServerDatasource, roomID string, evType, stateKey string) util.JSONResponse {
+func OnIncomingStateTypeRequest(req *http.Request, db storage.Database, roomID string, evType, stateKey string) util.JSONResponse {
 	// TODO(#287): Auth request and handle the case where the user has left (where
 	// we should return the state at the poin they left)
 

syncapi/storage/postgres/account_data_table.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package storage
+package postgres
 
 import (
 	"context"

syncapi/storage/postgres/current_room_state_table.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package storage
+package postgres
 
 import (
 	"context"

syncapi/storage/postgres/filtering.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package storage
+package postgres
 
 import (
 	"strings"

syncapi/storage/postgres/invites_table.go

@@ -1,4 +1,4 @@
-package storage
+package postgres
 
 import (
 	"context"

syncapi/storage/postgres/output_room_events_table.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package storage
+package postgres
 
 import (
 	"context"

syncapi/storage/postgres/syncserver.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package storage
+package postgres
 
 import (
 	"context"

syncapi/storage/storage.go

@@ -0,0 +1,63 @@
+// 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 storage
+
+import (
+	"context"
+	"errors"
+	"net/url"
+	"time"
+
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/common"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/storage/postgres"
+	"github.com/matrix-org/dendrite/syncapi/types"
+	"github.com/matrix-org/dendrite/typingserver/cache"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+type Database interface {
+	common.PartitionStorer
+	AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error)
+	Events(ctx context.Context, eventIDs []string) ([]gomatrixserverlib.Event, error)
+	WriteEvent(ctx context.Context, ev *gomatrixserverlib.Event, addStateEvents []gomatrixserverlib.Event, addStateEventIDs, removeStateEventIDs []string, transactionID *api.TransactionID) (pduPosition int64, returnErr error)
+	GetStateEvent(ctx context.Context, roomID, evType, stateKey string) (*gomatrixserverlib.Event, error)
+	GetStateEventsForRoom(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.FilterPart) (stateEvents []gomatrixserverlib.Event, err error)
+	SyncPosition(ctx context.Context) (types.SyncPosition, error)
+	IncrementalSync(ctx context.Context, device authtypes.Device, fromPos, toPos types.SyncPosition, numRecentEventsPerRoom int, wantFullState bool) (*types.Response, error)
+	CompleteSync(ctx context.Context, userID string, numRecentEventsPerRoom int) (*types.Response, error)
+	GetAccountDataInRange(ctx context.Context, userID string, oldPos, newPos int64, accountDataFilterPart *gomatrixserverlib.FilterPart) (map[string][]string, error)
+	UpsertAccountData(ctx context.Context, userID, roomID, dataType string) (int64, error)
+	AddInviteEvent(ctx context.Context, inviteEvent gomatrixserverlib.Event) (int64, error)
+	RetireInviteEvent(ctx context.Context, inviteEventID string) error
+	SetTypingTimeoutCallback(fn cache.TimeoutCallbackFn)
+	AddTypingUser(userID, roomID string, expireTime *time.Time) int64
+	RemoveTypingUser(userID, roomID string) int64
+}
+
+// NewPublicRoomsServerDatabase opens a database connection.
+func NewSyncServerDatasource(dataSourceName string) (Database, error) {
+	uri, err := url.Parse(dataSourceName)
+	if err != nil {
+		return nil, err
+	}
+	switch uri.Scheme {
+	case "postgres":
+		return postgres.NewSyncServerDatasource(dataSourceName)
+	default:
+		return nil, errors.New("unknown schema")
+	}
+}

syncapi/sync/notifier.go

@@ -141,7 +141,7 @@ func (n *Notifier) GetListener(req syncRequest) UserStreamListener {
 }
 
 // Load the membership states required to notify users correctly.
-func (n *Notifier) Load(ctx context.Context, db *storage.SyncServerDatasource) error {
+func (n *Notifier) Load(ctx context.Context, db storage.Database) error {
 	roomToUsers, err := db.AllJoinedUsersInRooms(ctx)
 	if err != nil {
 		return err

syncapi/sync/requestpool.go

@@ -31,13 +31,13 @@ import (
 
 // RequestPool manages HTTP long-poll connections for /sync
 type RequestPool struct {
-	db        *storage.SyncServerDatasource
+	db        storage.Database
 	accountDB *accounts.Database
 	notifier  *Notifier
 }
 
 // NewRequestPool makes a new RequestPool
-func NewRequestPool(db *storage.SyncServerDatasource, n *Notifier, adb *accounts.Database) *RequestPool {
+func NewRequestPool(db storage.Database, n *Notifier, adb *accounts.Database) *RequestPool {
 	return &RequestPool{db, adb, n}
 }