diff --git a/syncapi/storage/postgres/receipt_table.go b/syncapi/storage/postgres/receipt_table.go index 9fe748568..23c66910f 100644 --- a/syncapi/storage/postgres/receipt_table.go +++ b/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 +} diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index c0ae3d7a9..71a42003e 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/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()), + PDUPosition: types.StreamPosition(maxEventID), + TypingPosition: types.StreamPosition(d.EDUCache.GetLatestSyncPosition()), + ReceiptPosition: types.StreamPosition(maxReceiptID), } return } diff --git a/syncapi/storage/sqlite3/receipt_table.go b/syncapi/storage/sqlite3/receipt_table.go index 427037bb9..dfde1fd2d 100644 --- a/syncapi/storage/sqlite3/receipt_table.go +++ b/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 +} diff --git a/syncapi/storage/sqlite3/stream_id_table.go b/syncapi/storage/sqlite3/stream_id_table.go index b3fc264c3..f73be422d 100644 --- a/syncapi/storage/sqlite3/stream_id_table.go +++ b/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 = "" + diff --git a/syncapi/storage/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go index f07359e7a..1ad0e9473 100644 --- a/syncapi/storage/sqlite3/syncserver.go +++ b/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 diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index a2d8791b6..7a166d439 100644 --- a/syncapi/storage/tables/interface.go +++ b/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) }