Add max receipt queries, missing stream_id table entry for SQLite

syncapi/storage/postgres/receipt_table.go

@@ -59,10 +59,14 @@ const selectRoomReceipts = "" +
 	" FROM syncapi_receipts" +
 	" WHERE room_id = ANY($1) AND id > $2"
 
+const selectMaxReceiptIDSQL = "" +
+	"SELECT MAX(id) FROM syncapi_receipts"
+
 type receiptStatements struct {
 	db                 *sql.DB
 	upsertReceipt      *sql.Stmt
 	selectRoomReceipts *sql.Stmt
+	selectMaxReceiptID *sql.Stmt
 }
 
 func NewPostgresReceiptsTable(db *sql.DB) (tables.Receipts, error) {
@@ -79,6 +83,9 @@ func NewPostgresReceiptsTable(db *sql.DB) (tables.Receipts, error) {
 	if r.selectRoomReceipts, err = db.Prepare(selectRoomReceipts); err != nil {
 		return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err)
 	}
+	if r.selectMaxReceiptID, err = db.Prepare(selectMaxReceiptIDSQL); err != nil {
+		return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err)
+	}
 	return r, nil
 }
 
@@ -105,3 +112,15 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs
 	}
 	return res, rows.Err()
 }
+
+func (s *receiptStatements) SelectMaxReceiptID(
+	ctx context.Context, txn *sql.Tx,
+) (id int64, err error) {
+	var nullableID sql.NullInt64
+	stmt := sqlutil.TxStmt(txn, s.selectMaxReceiptID)
+	err = stmt.QueryRowContext(ctx).Scan(&nullableID)
+	if nullableID.Valid {
+		id = nullableID.Int64
+	}
+	return
+}

syncapi/storage/shared/syncserver.go

@@ -483,10 +483,15 @@ func (d *Database) syncPositionTx(
 	if maxPeekID > maxEventID {
 		maxEventID = maxPeekID
 	}
+	maxReceiptID, err := d.Receipts.SelectMaxReceiptID(ctx, txn)
+	if err != nil {
+		return sp, err
+	}
 	// TODO: complete these positions
 	sp = types.StreamingToken{
 		PDUPosition:     types.StreamPosition(maxEventID),
 		TypingPosition:  types.StreamPosition(d.EDUCache.GetLatestSyncPosition()),
+		ReceiptPosition: types.StreamPosition(maxReceiptID),
 	}
 	return
 }

syncapi/storage/sqlite3/receipt_table.go

@@ -55,11 +55,15 @@ const selectRoomReceipts = "" +
 	" FROM syncapi_receipts" +
 	" WHERE id > $1 and room_id in ($2)"
 
+const selectMaxReceiptIDSQL = "" +
+	"SELECT MAX(id) FROM syncapi_receipts"
+
 type receiptStatements struct {
 	db                 *sql.DB
 	streamIDStatements *streamIDStatements
 	upsertReceipt      *sql.Stmt
 	selectRoomReceipts *sql.Stmt
+	selectMaxReceiptID *sql.Stmt
 }
 
 func NewSqliteReceiptsTable(db *sql.DB, streamID *streamIDStatements) (tables.Receipts, error) {
@@ -77,6 +81,9 @@ func NewSqliteReceiptsTable(db *sql.DB, streamID *streamIDStatements) (tables.Re
 	if r.selectRoomReceipts, err = db.Prepare(selectRoomReceipts); err != nil {
 		return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err)
 	}
+	if r.selectMaxReceiptID, err = db.Prepare(selectMaxReceiptIDSQL); err != nil {
+		return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err)
+	}
 	return r, nil
 }
 
@@ -116,3 +123,15 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs
 	}
 	return res, rows.Err()
 }
+
+func (s *receiptStatements) SelectMaxReceiptID(
+	ctx context.Context, txn *sql.Tx,
+) (id int64, err error) {
+	var nullableID sql.NullInt64
+	stmt := sqlutil.TxStmt(txn, s.selectMaxReceiptID)
+	err = stmt.QueryRowContext(ctx).Scan(&nullableID)
+	if nullableID.Valid {
+		id = nullableID.Int64
+	}
+	return
+}

syncapi/storage/sqlite3/stream_id_table.go

@@ -18,6 +18,8 @@ CREATE TABLE IF NOT EXISTS syncapi_stream_id (
 );
 INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("global", 0)
   ON CONFLICT DO NOTHING;
+INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("receipt", 0)
+  ON CONFLICT DO NOTHING;
 `
 
 const increaseStreamIDStmt = "" +

syncapi/storage/sqlite3/syncserver.go

@@ -53,6 +53,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
 	return &d, nil
 }
 
+// nolint:gocyclo
 func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (err error) {
 	if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "syncapi"); err != nil {
 		return err

syncapi/storage/tables/interface.go

@@ -161,4 +161,5 @@ type Filter interface {
 type Receipts interface {
 	UpsertReceipt(ctx context.Context, txn *sql.Tx, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error)
 	SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]eduAPI.OutputReceiptEvent, error)
+	SelectMaxReceiptID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }