hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-08-02 22:22:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add (broken) postgres; advance streampos whenever sync output changes

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-08 00:56:50 +01:00
parent 843b7a7d04
commit 55c7f2c892
5 changed files with 252 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

48
syncapi/storage/sqlite3/peeks_table.go
View file

@ -32,6 +32,7 @@ CREATE TABLE IF NOT EXISTS syncapi_peeks (
user_id TEXT NOT NULL,
device_id TEXT NOT NULL,
new BOOL NOT NULL DEFAULT true,
deleted BOOL NOT NULL DEFAULT false,
-- When the peek was created in UNIX epoch ms.
creation_ts INTEGER NOT NULL
);
@ -42,23 +43,26 @@ CREATE INDEX IF NOT EXISTS syncapi_peeks_user_id_device_id_idx ON syncapi_peeks(
const insertPeekSQL = "" +
"INSERT INTO syncapi_peeks" +
" (room_id, user_id, device_id, creation_ts)" +
" VALUES ($1, $2, $3, $4)"
" (id, room_id, user_id, device_id, creation_ts)" +
" VALUES ($1, $2, $3, $4, $5)"
const deletePeekSQL = "" +
"DELETE FROM syncapi_peeks WHERE room_id = $1 AND user_id = $2 and device_id = $3"
"UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3 AND device_id = $4"
const deletePeeksSQL = "" +
"DELETE FROM syncapi_peeks WHERE room_id = $1 AND user_id = $2"
"UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3"
const selectPeeksSQL = "" +
"SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 and device_id = $2"
"SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND deleted=false"
const selectPeekingDevicesSQL = "" +
"SELECT room_id, user_id, device_id FROM syncapi_peeks"
"SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false"
const markPeeksAsOldSQL = "" +
"UPDATE syncapi_peeks SET new=false WHERE user_id = $1 and device_id = $2"
"UPDATE syncapi_peeks SET new=false, id=$1 WHERE user_id = $2 AND device_id = $3 AND deleted=false"
const selectMaxPeekIDSQL = "" +
"SELECT MAX(id) FROM syncapi_peeks"
type peekStatements struct {
db *sql.DB
@ -69,6 +73,7 @@ type peekStatements struct {
selectPeeksStmt *sql.Stmt
selectPeekingDevicesStmt *sql.Stmt
markPeeksAsOldStmt *sql.Stmt
selectMaxPeekIDStmt *sql.Stmt
}
func NewSqlitePeeksTable(db *sql.DB, streamID *streamIDStatements) (tables.Peeks, error) {
@ -98,6 +103,9 @@ func NewSqlitePeeksTable(db *sql.DB, streamID *streamIDStatements) (tables.Peeks
if s.markPeeksAsOldStmt, err = db.Prepare(markPeeksAsOldSQL); err != nil {
return nil, err
}
if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil {
return nil, err
}
return s, nil
}
@ -109,7 +117,7 @@ func (s *peekStatements) InsertPeek(
return
}
nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
_, err = sqlutil.TxStmt(txn, s.insertPeekStmt).ExecContext(ctx, roomID, userID, deviceID, nowMilli)
_, err = sqlutil.TxStmt(txn, s.insertPeekStmt).ExecContext(ctx, streamPos, roomID, userID, deviceID, nowMilli)
return
}
@ -120,7 +128,7 @@ func (s *peekStatements) DeletePeek(
if err != nil {
return
}
_, err = sqlutil.TxStmt(txn, s.deletePeekStmt).ExecContext(ctx, roomID, userID, deviceID)
_, err = sqlutil.TxStmt(txn, s.deletePeekStmt).ExecContext(ctx, streamPos, roomID, userID, deviceID)
return
}
@ -131,7 +139,7 @@ func (s *peekStatements) DeletePeeks(
if err != nil {
return
}
_, err = sqlutil.TxStmt(txn, s.deletePeeksStmt).ExecContext(ctx, roomID, userID)
_, err = sqlutil.TxStmt(txn, s.deletePeeksStmt).ExecContext(ctx, streamPos, roomID, userID)
return
}
@ -157,8 +165,12 @@ func (s *peekStatements) SelectPeeks(
func (s *peekStatements) MarkPeeksAsOld(
ctx context.Context, txn *sql.Tx, userID, deviceID string,
) (err error) {
_, err = sqlutil.TxStmt(txn, s.markPeeksAsOldStmt).ExecContext(ctx, userID, deviceID)
) (streamPos types.StreamPosition, err error) {
streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return
}
_, err = sqlutil.TxStmt(txn, s.markPeeksAsOldStmt).ExecContext(ctx, streamPos, userID, deviceID)
return
}
@ -183,3 +195,15 @@ func (s *peekStatements) SelectPeekingDevices(
}
return result, nil
}
func (s *peekStatements) SelectMaxPeekID(
ctx context.Context, txn *sql.Tx,
) (id int64, err error) {
var nullableID sql.NullInt64
stmt := sqlutil.TxStmt(txn, s.selectMaxPeekIDStmt)
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
if nullableID.Valid {
id = nullableID.Int64
}
return
}