Write OutputRoomEvents into the database and remember the arrival order
ADD
src/github.com/matrix-org/dendrite/clientapi/storage/output_room_events_table.go
+package storage
+
+import (
+ "database/sql"
+
+ "github.com/lib/pq"
+)
+
+const outputRoomEventsSchema = `
+-- Stores output room events received from the roomserver.
+CREATE TABLE IF NOT EXISTS output_room_events (
+ -- An incrementing ID which denotes the position in the log that this event resides at.
+ -- NB: 'serial' makes no guarantees to increment by 1 every time, only that it increments.
+ -- This isn't a problem for us since we just want to order by this field.
+ id BIGSERIAL PRIMARY KEY,
+ -- The 'room_id' key for the event.
+ room_id TEXT NOT NULL,
+ -- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
+ event_json TEXT NOT NULL,
+ -- A list of event IDs which represent a delta of added/removed room state.
+ add_state_ids TEXT[] NOT NULL,
+ remove_state_ids TEXT[] NOT NULL
+);
+`
+
+const insertEventSQL = "" +
+ "INSERT INTO output_room_events (room_id, event_json, add_state_ids, remove_state_ids) VALUES ($1, $2, $3, $4)"
+
+type outputRoomEventsStatements struct {
+ insertEventStmt *sql.Stmt
+}
+
+func (s *outputRoomEventsStatements) prepare(db *sql.DB) (err error) {
+ _, err = db.Exec(outputRoomEventsSchema)
+ if err != nil {
+ return
+ }
+ if s.insertEventStmt, err = db.Prepare(insertEventSQL); err != nil {
+ return
+ }
+ return
+}
+
+// InsertEvent into the output_room_events table. addState and removeState are an optional list of state event IDs.
+func (s *outputRoomEventsStatements) InsertEvent(roomID string, eventJSON []byte, addState, removeState []string) error {
+ _, err := s.insertEventStmt.Exec(roomID, eventJSON, pq.StringArray(addState), pq.StringArray(removeState))
+ return err
+}

MODIFY
src/github.com/matrix-org/dendrite/clientapi/storage/syncserver.go
src/github.com/matrix-org/dendrite/clientapi/storage/syncserver.go
@@ -2,16 +2,17 @@ package storage
import (
"database/sql"
-
- "github.com/matrix-org/dendrite/common"
// Import the postgres database driver.
_ "github.com/lib/pq"
+ "github.com/matrix-org/dendrite/common"
+ "github.com/matrix-org/gomatrixserverlib"
)
// SyncServerDatabase represents a sync server database
type SyncServerDatabase struct {
db *sql.DB
partitions common.PartitionOffsetStatements
+ events outputRoomEventsStatements
}
// NewSyncServerDatabase creates a new sync server database
@@ -25,7 +26,17 @@ func NewSyncServerDatabase(dataSourceName string) (*SyncServerDatabase, error) {
if err = partitions.Prepare(db); err != nil {
return nil, err
}
- return &SyncServerDatabase{db, partitions}, nil
+ events := outputRoomEventsStatements{}
+ if err = events.prepare(db); err != nil {
+ return nil, err
+ }
+ return &SyncServerDatabase{db, partitions, events}, nil
+}
+
+// WriteEvent into the database. It is not safe to call this function from multiple goroutines, as it would create races
+// when generating the stream position for this event. Returns an error if there was a problem inserting this event.
+func (d *SyncServerDatabase) WriteEvent(ev *gomatrixserverlib.Event, addStateEventIDs, removeStateEventIDs []string) error {
+ return d.events.InsertEvent(ev.RoomID(), ev.JSON(), addStateEventIDs, removeStateEventIDs)
}
// PartitionOffsets implements common.PartitionStorer

MODIFY
src/github.com/matrix-org/dendrite/clientapi/sync/syncserver.go
src/github.com/matrix-org/dendrite/clientapi/sync/syncserver.go
package sync
import (
+ "encoding/json"
+
log "github.com/Sirupsen/logrus"
"github.com/matrix-org/dendrite/clientapi/config"
+ "github.com/matrix-org/dendrite/clientapi/storage"
"github.com/matrix-org/dendrite/common"
+ "github.com/matrix-org/dendrite/roomserver/api"
+ "github.com/matrix-org/gomatrixserverlib"
sarama "gopkg.in/Shopify/sarama.v1"
)
// Server contains all the logic for running a sync server
type Server struct {
roomServerConsumer *common.ContinualConsumer
+ db *storage.SyncServerDatabase
}
// NewServer creates a new sync server. Call Start() to begin consuming from room servers.
-func NewServer(cfg *config.Sync, store common.PartitionStorer) (*Server, error) {
+func NewServer(cfg *config.Sync, store *storage.SyncServerDatabase) (*Server, error) {
kafkaConsumer, err := sarama.NewConsumer(cfg.KafkaConsumerURIs, nil)
if err != nil {
return nil, err
@@ -26,6 +32,7 @@ func NewServer(cfg *config.Sync, store common.PartitionStorer) (*Server, error)
}
s := &Server{
roomServerConsumer: &consumer,
+ db: store,
}
consumer.ProcessMessage = s.onMessage
@@ -38,6 +45,29 @@ func (s *Server) Start() error {
}
func (s *Server) onMessage(msg *sarama.ConsumerMessage) error {
- log.WithField("key", string(msg.Key)).WithField("val", string(msg.Value)).Info("Recv")
+ // Parse out the event JSON
+ var output api.OutputRoomEvent
+ if err := json.Unmarshal(msg.Value, &output); err != nil {
+ // If the message was invalid, log it and move on to the next message in the stream
+ log.WithError(err).Errorf("roomserver output log: message parse failure")
+ return nil
+ }
+
+ ev, err := gomatrixserverlib.NewEventFromTrustedJSON(output.Event, false)
+ if err != nil {
+ log.WithError(err).Errorf("roomserver output log: event parse failure")
+ return nil
+ }
+ log.WithFields(log.Fields{
+ "event_id": ev.EventID(),
+ "room_id": ev.RoomID(),
+ }).Info("received event from roomserver")
+
+ if err := s.db.WriteEvent(&ev, output.AddsStateEventIDs, output.RemovesStateEventIDs); err != nil {
+ // panic rather than continue with an inconsistent database
+ log.WithError(err).WithField("OutputRoomEvent", output).Panicf("roomserver output log: write event failure")
+ return nil
+ }
+
return nil
}

Write OutputRoomEvents into the database and remember the arrival order (#54)