CS API: Support for /messages, fixes for /sync (#847)

* Merge forward

* Tidy up a bit

* TODO: What to do with NextBatch here?

* Replace SyncPosition with PaginationToken throughout syncapi

* Fix PaginationTokens

* Fix lint errors

* Add a couple of missing functions into the syncapi external storage interface

* Some updates based on review comments from @babolivier

* Some updates based on review comments from @babolivier

* argh whitespacing

* Fix opentracing span

* Remove dead code

* Don't overshadow err (fix lint issue)

* Handle extremities after inserting event into topology

* Try insert event topology as ON CONFLICT DO NOTHING

* Prevent OOB error in addRoomDeltaToResponse

* Thwarted by gocyclo again

* Fix NewPaginationTokenFromString, define unit test for it

* Update pagination token test

* Update sytest-whitelist

* Hopefully fix some of the sync batch tokens

* Remove extraneous sync position func

* Revert to topology tokens in addRoomDeltaToResponse etc

* Fix typo

* Remove prevPDUPos as dead now that backwardTopologyPos is used instead

* Fix selectEventsWithEventIDsSQL

* Update sytest-blacklist

* Update sytest-whitelist

1 files changed, 118 insertions, 0 deletions

diff --git a/syncapi/storage/postgres/backward_extremities_table.go b/syncapi/storage/postgres/backward_extremities_table.go
new file mode 100644
index 00000000..476d26fa
--- /dev/null
+++ b/syncapi/storage/postgres/backward_extremities_table.go
@@ -0,0 +1,118 @@
+// Copyright 2018 New Vector 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 postgres
+
+import (
+	"context"
+	"database/sql"
+)
+
+const backwardExtremitiesSchema = `
+-- Stores output room events received from the roomserver.
+CREATE TABLE IF NOT EXISTS syncapi_backward_extremities (
+	-- The 'room_id' key for the event.
+	room_id TEXT NOT NULL,
+	-- The event ID for the event.
+	event_id TEXT NOT NULL,
+
+	PRIMARY KEY(room_id, event_id)
+);
+`
+
+const insertBackwardExtremitySQL = "" +
+	"INSERT INTO syncapi_backward_extremities (room_id, event_id)" +
+	" VALUES ($1, $2)"
+
+const selectBackwardExtremitiesForRoomSQL = "" +
+	"SELECT event_id FROM syncapi_backward_extremities WHERE room_id = $1"
+
+const isBackwardExtremitySQL = "" +
+	"SELECT EXISTS (" +
+	" SELECT TRUE FROM syncapi_backward_extremities" +
+	" WHERE room_id = $1 AND event_id = $2" +
+	")"
+
+const deleteBackwardExtremitySQL = "" +
+	"DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND event_id = $2"
+
+type backwardExtremitiesStatements struct {
+	insertBackwardExtremityStmt          *sql.Stmt
+	selectBackwardExtremitiesForRoomStmt *sql.Stmt
+	isBackwardExtremityStmt              *sql.Stmt
+	deleteBackwardExtremityStmt          *sql.Stmt
+}
+
+func (s *backwardExtremitiesStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(backwardExtremitiesSchema)
+	if err != nil {
+		return
+	}
+	if s.insertBackwardExtremityStmt, err = db.Prepare(insertBackwardExtremitySQL); err != nil {
+		return
+	}
+	if s.selectBackwardExtremitiesForRoomStmt, err = db.Prepare(selectBackwardExtremitiesForRoomSQL); err != nil {
+		return
+	}
+	if s.isBackwardExtremityStmt, err = db.Prepare(isBackwardExtremitySQL); err != nil {
+		return
+	}
+	if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil {
+		return
+	}
+	return
+}
+
+func (s *backwardExtremitiesStatements) insertsBackwardExtremity(
+	ctx context.Context, roomID, eventID string,
+) (err error) {
+	_, err = s.insertBackwardExtremityStmt.ExecContext(ctx, roomID, eventID)
+	return
+}
+
+func (s *backwardExtremitiesStatements) selectBackwardExtremitiesForRoom(
+	ctx context.Context, roomID string,
+) (eventIDs []string, err error) {
+	eventIDs = make([]string, 0)
+
+	rows, err := s.selectBackwardExtremitiesForRoomStmt.QueryContext(ctx, roomID)
+	if err != nil {
+		return
+	}
+
+	for rows.Next() {
+		var eID string
+		if err = rows.Scan(&eID); err != nil {
+			return
+		}
+
+		eventIDs = append(eventIDs, eID)
+	}
+
+	return
+}
+
+func (s *backwardExtremitiesStatements) isBackwardExtremity(
+	ctx context.Context, roomID, eventID string,
+) (isBE bool, err error) {
+	err = s.isBackwardExtremityStmt.QueryRowContext(ctx, roomID, eventID).Scan(&isBE)
+	return
+}
+
+func (s *backwardExtremitiesStatements) deleteBackwardExtremity(
+	ctx context.Context, roomID, eventID string,
+) (err error) {
+	_, err = s.insertBackwardExtremityStmt.ExecContext(ctx, roomID, eventID)
+	return
+}