mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-06 22:43:10 -06:00
Add storage layer
This commit is contained in:
parent
e98d6848e3
commit
3bbe097162
|
|
@ -30,6 +30,7 @@ import (
|
||||||
|
|
||||||
type Database interface {
|
type Database interface {
|
||||||
UserRoomKeys
|
UserRoomKeys
|
||||||
|
ReportedEvents
|
||||||
// Do we support processing input events for more than one room at a time?
|
// Do we support processing input events for more than one room at a time?
|
||||||
SupportsConcurrentRoomInputs() bool
|
SupportsConcurrentRoomInputs() bool
|
||||||
AssignRoomNID(ctx context.Context, roomID spec.RoomID, roomVersion gomatrixserverlib.RoomVersion) (roomNID types.RoomNID, err error)
|
AssignRoomNID(ctx context.Context, roomID spec.RoomID, roomVersion gomatrixserverlib.RoomVersion) (roomNID types.RoomNID, err error)
|
||||||
|
|
@ -257,3 +258,11 @@ type EventDatabase interface {
|
||||||
) (gomatrixserverlib.PDU, gomatrixserverlib.PDU, error)
|
) (gomatrixserverlib.PDU, gomatrixserverlib.PDU, error)
|
||||||
StoreEvent(ctx context.Context, event gomatrixserverlib.PDU, roomInfo *types.RoomInfo, eventTypeNID types.EventTypeNID, eventStateKeyNID types.EventStateKeyNID, authEventNIDs []types.EventNID, isRejected bool) (types.EventNID, types.StateAtEvent, error)
|
StoreEvent(ctx context.Context, event gomatrixserverlib.PDU, roomInfo *types.RoomInfo, eventTypeNID types.EventTypeNID, eventStateKeyNID types.EventStateKeyNID, authEventNIDs []types.EventNID, isRejected bool) (types.EventNID, types.StateAtEvent, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type ReportedEvents interface {
|
||||||
|
InsertReportedEvent(
|
||||||
|
ctx context.Context,
|
||||||
|
roomID, eventID, reportingUserID, reason string,
|
||||||
|
score int64,
|
||||||
|
) (int64, error)
|
||||||
|
}
|
||||||
|
|
|
||||||
85
roomserver/storage/postgres/reported_events_table.go
Normal file
85
roomserver/storage/postgres/reported_events_table.go
Normal file
|
|
@ -0,0 +1,85 @@
|
||||||
|
// Copyright 2023 The Matrix.org Foundation C.I.C.
|
||||||
|
//
|
||||||
|
// 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"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/types"
|
||||||
|
"github.com/matrix-org/gomatrixserverlib/spec"
|
||||||
|
)
|
||||||
|
|
||||||
|
const reportedEventsScheme = `
|
||||||
|
CREATE SEQUENCE IF NOT EXISTS roomserver_reported_events_id_seq;
|
||||||
|
CREATE TABLE IF NOT EXISTS roomserver_reported_events
|
||||||
|
(
|
||||||
|
id BIGINT PRIMARY KEY DEFAULT nextval('roomserver_reported_events_id_seq'),
|
||||||
|
room_nid BIGINT NOT NULL,
|
||||||
|
event_nid BIGINT NOT NULL,
|
||||||
|
user_id TEXT NOT NULL,
|
||||||
|
reason TEXT,
|
||||||
|
score INTEGER,
|
||||||
|
received_ts BIGINT NOT NULL
|
||||||
|
);`
|
||||||
|
|
||||||
|
const insertReportedEventSQL = `
|
||||||
|
INSERT INTO roomserver_reported_events (room_nid, event_nid, user_id, reason, score, received_ts)
|
||||||
|
VALUES ($1, $2, $3, $4, $5, $6)
|
||||||
|
RETURNING id
|
||||||
|
`
|
||||||
|
|
||||||
|
type reportedEventsStatements struct {
|
||||||
|
insertReportedEventsStmt *sql.Stmt
|
||||||
|
}
|
||||||
|
|
||||||
|
func CreateReportedEventsTable(db *sql.DB) error {
|
||||||
|
_, err := db.Exec(reportedEventsScheme)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func PrepareReportEventsTable(db *sql.DB) (tables.ReportedEvents, error) {
|
||||||
|
s := &reportedEventsStatements{}
|
||||||
|
|
||||||
|
return s, sqlutil.StatementList{
|
||||||
|
{&s.insertReportedEventsStmt, insertReportedEventSQL},
|
||||||
|
}.Prepare(db)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *reportedEventsStatements) InsertReportedEvent(
|
||||||
|
ctx context.Context,
|
||||||
|
txn *sql.Tx,
|
||||||
|
roomNID types.RoomNID,
|
||||||
|
eventNID types.EventNID,
|
||||||
|
reportingUserID string,
|
||||||
|
reason string,
|
||||||
|
score int64,
|
||||||
|
) (int64, error) {
|
||||||
|
stmt := sqlutil.TxStmt(txn, r.insertReportedEventsStmt)
|
||||||
|
|
||||||
|
var reportID int64
|
||||||
|
err := stmt.QueryRowContext(ctx,
|
||||||
|
roomNID,
|
||||||
|
eventNID,
|
||||||
|
reportingUserID,
|
||||||
|
reason,
|
||||||
|
score,
|
||||||
|
spec.AsTimestamp(time.Now()),
|
||||||
|
).Scan(&reportID)
|
||||||
|
return reportID, err
|
||||||
|
}
|
||||||
|
|
@ -134,6 +134,9 @@ func (d *Database) create(db *sql.DB) error {
|
||||||
if err := CreateUserRoomKeysTable(db); err != nil {
|
if err := CreateUserRoomKeysTable(db); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
if err := CreateReportedEventsTable(db); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -199,6 +202,10 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
reportedEvents, err := PrepareReportEventsTable(db)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
d.Database = shared.Database{
|
d.Database = shared.Database{
|
||||||
DB: db,
|
DB: db,
|
||||||
|
|
@ -212,6 +219,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
|
||||||
EventStateKeysTable: eventStateKeys,
|
EventStateKeysTable: eventStateKeys,
|
||||||
PrevEventsTable: prevEvents,
|
PrevEventsTable: prevEvents,
|
||||||
RedactionsTable: redactions,
|
RedactionsTable: redactions,
|
||||||
|
ReportedEventsTable: reportedEvents,
|
||||||
},
|
},
|
||||||
Cache: cache,
|
Cache: cache,
|
||||||
Writer: writer,
|
Writer: writer,
|
||||||
|
|
|
||||||
|
|
@ -61,6 +61,7 @@ type EventDatabase struct {
|
||||||
EventStateKeysTable tables.EventStateKeys
|
EventStateKeysTable tables.EventStateKeys
|
||||||
PrevEventsTable tables.PreviousEvents
|
PrevEventsTable tables.PreviousEvents
|
||||||
RedactionsTable tables.Redactions
|
RedactionsTable tables.Redactions
|
||||||
|
ReportedEventsTable tables.ReportedEvents
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Database) SupportsConcurrentRoomInputs() bool {
|
func (d *Database) SupportsConcurrentRoomInputs() bool {
|
||||||
|
|
@ -1882,6 +1883,40 @@ func (d *Database) SelectUserIDsForPublicKeys(ctx context.Context, publicKeys ma
|
||||||
return result, err
|
return result, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// InsertReportedEvent stores a reported event.
|
||||||
|
func (d *Database) InsertReportedEvent(
|
||||||
|
ctx context.Context,
|
||||||
|
roomID, eventID, reportingUserID, reason string,
|
||||||
|
score int64,
|
||||||
|
) (int64, error) {
|
||||||
|
roomInfo, err := d.roomInfo(ctx, nil, roomID)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
if roomInfo == nil {
|
||||||
|
return 0, fmt.Errorf("room does not exist")
|
||||||
|
}
|
||||||
|
|
||||||
|
events, err := d.eventsFromIDs(ctx, nil, roomInfo, []string{eventID}, NoFilter)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
if len(events) == 0 {
|
||||||
|
return 0, fmt.Errorf("unable to find requested event")
|
||||||
|
}
|
||||||
|
|
||||||
|
var reportID int64
|
||||||
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
||||||
|
reportID, err = d.ReportedEventsTable.InsertReportedEvent(ctx, txn, roomInfo.RoomNID, events[0].EventNID, reportingUserID, reason, score)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
return reportID, err
|
||||||
|
}
|
||||||
|
|
||||||
// FIXME TODO: Remove all this - horrible dupe with roomserver/state. Can't use the original impl because of circular loops
|
// FIXME TODO: Remove all this - horrible dupe with roomserver/state. Can't use the original impl because of circular loops
|
||||||
// it should live in this package!
|
// it should live in this package!
|
||||||
|
|
||||||
|
|
|
||||||
85
roomserver/storage/sqlite3/reported_events_table.go
Normal file
85
roomserver/storage/sqlite3/reported_events_table.go
Normal file
|
|
@ -0,0 +1,85 @@
|
||||||
|
// Copyright 2023 The Matrix.org Foundation C.I.C.
|
||||||
|
//
|
||||||
|
// 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 sqlite3
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/types"
|
||||||
|
"github.com/matrix-org/gomatrixserverlib/spec"
|
||||||
|
)
|
||||||
|
|
||||||
|
const reportedEventsScheme = `
|
||||||
|
CREATE SEQUENCE IF NOT EXISTS roomserver_reported_events_id_seq;
|
||||||
|
CREATE TABLE IF NOT EXISTS roomserver_reported_events
|
||||||
|
(
|
||||||
|
id BIGINT PRIMARY KEY DEFAULT nextval('roomserver_reported_events_id_seq'),
|
||||||
|
room_nid BIGINT NOT NULL,
|
||||||
|
event_nid BIGINT NOT NULL,
|
||||||
|
user_id TEXT NOT NULL,
|
||||||
|
reason TEXT,
|
||||||
|
score INTEGER,
|
||||||
|
received_ts BIGINT NOT NULL
|
||||||
|
);`
|
||||||
|
|
||||||
|
const insertReportedEventSQL = `
|
||||||
|
INSERT INTO roomserver_reported_events (room_nid, event_nid, user_id, reason, score, received_ts)
|
||||||
|
VALUES ($1, $2, $3, $4, $5, $6)
|
||||||
|
RETURNING id
|
||||||
|
`
|
||||||
|
|
||||||
|
type reportedEventsStatements struct {
|
||||||
|
insertReportedEventsStmt *sql.Stmt
|
||||||
|
}
|
||||||
|
|
||||||
|
func CreateReportedEventsTable(db *sql.DB) error {
|
||||||
|
_, err := db.Exec(reportedEventsScheme)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func PrepareReportEventsTable(db *sql.DB) (tables.ReportedEvents, error) {
|
||||||
|
s := &reportedEventsStatements{}
|
||||||
|
|
||||||
|
return s, sqlutil.StatementList{
|
||||||
|
{&s.insertReportedEventsStmt, insertReportedEventSQL},
|
||||||
|
}.Prepare(db)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *reportedEventsStatements) InsertReportedEvent(
|
||||||
|
ctx context.Context,
|
||||||
|
txn *sql.Tx,
|
||||||
|
roomNID types.RoomNID,
|
||||||
|
eventNID types.EventNID,
|
||||||
|
reportingUserID string,
|
||||||
|
reason string,
|
||||||
|
score int64,
|
||||||
|
) (int64, error) {
|
||||||
|
stmt := sqlutil.TxStmt(txn, r.insertReportedEventsStmt)
|
||||||
|
|
||||||
|
var reportID int64
|
||||||
|
err := stmt.QueryRowContext(ctx,
|
||||||
|
roomNID,
|
||||||
|
eventNID,
|
||||||
|
reportingUserID,
|
||||||
|
reason,
|
||||||
|
score,
|
||||||
|
spec.AsTimestamp(time.Now()),
|
||||||
|
).Scan(&reportID)
|
||||||
|
return reportID, err
|
||||||
|
}
|
||||||
|
|
@ -141,7 +141,9 @@ func (d *Database) create(db *sql.DB) error {
|
||||||
if err := CreateUserRoomKeysTable(db); err != nil {
|
if err := CreateUserRoomKeysTable(db); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
if err := CreateReportedEventsTable(db); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -206,6 +208,10 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
reportedEvents, err := PrepareReportEventsTable(db)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
d.Database = shared.Database{
|
d.Database = shared.Database{
|
||||||
DB: db,
|
DB: db,
|
||||||
|
|
@ -219,6 +225,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
|
||||||
EventJSONTable: eventJSON,
|
EventJSONTable: eventJSON,
|
||||||
PrevEventsTable: prevEvents,
|
PrevEventsTable: prevEvents,
|
||||||
RedactionsTable: redactions,
|
RedactionsTable: redactions,
|
||||||
|
ReportedEventsTable: reportedEvents,
|
||||||
},
|
},
|
||||||
Cache: cache,
|
Cache: cache,
|
||||||
Writer: writer,
|
Writer: writer,
|
||||||
|
|
|
||||||
|
|
@ -127,6 +127,18 @@ type Invites interface {
|
||||||
SelectInviteActiveForUserInRoom(ctx context.Context, txn *sql.Tx, targetUserNID types.EventStateKeyNID, roomNID types.RoomNID) ([]types.EventStateKeyNID, []string, []byte, error)
|
SelectInviteActiveForUserInRoom(ctx context.Context, txn *sql.Tx, targetUserNID types.EventStateKeyNID, roomNID types.RoomNID) ([]types.EventStateKeyNID, []string, []byte, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type ReportedEvents interface {
|
||||||
|
InsertReportedEvent(
|
||||||
|
ctx context.Context,
|
||||||
|
txn *sql.Tx,
|
||||||
|
roomNID types.RoomNID,
|
||||||
|
eventNID types.EventNID,
|
||||||
|
reportingUserID string,
|
||||||
|
reason string,
|
||||||
|
score int64,
|
||||||
|
) (int64, error)
|
||||||
|
}
|
||||||
|
|
||||||
type MembershipState int64
|
type MembershipState int64
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue