Merge pull request #38 from matrix-org/kegan/sync-on-startup

Start client sync streams on startup

src/github.com/matrix-org/go-neb/clients/clients.go

@@ -43,6 +43,22 @@ func (c *Clients) Update(config types.ClientConfig) (types.ClientConfig, error)
 	return old.config, err
 }
 
+// Start listening on client /sync streams
+func (c *Clients) Start() error {
+	configs, err := c.db.LoadMatrixClientConfigs()
+	if err != nil {
+		return err
+	}
+	for _, cfg := range configs {
+		if cfg.Sync {
+			if _, err := c.Client(cfg.UserID); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 type clientEntry struct {
 	config types.ClientConfig
 	client *matrix.Client
@@ -139,6 +155,33 @@ func (c *Clients) newClient(config types.ClientConfig) (*matrix.Client, error) {
 		plugin.OnMessage(plugins, client, event)
 	})
 
+	if config.AutoJoinRooms {
+		client.Worker.OnEventType("m.room.member", func(event *matrix.Event) {
+			if event.StateKey != config.UserID {
+				return // not our member event
+			}
+			m := event.Content["membership"]
+			membership, ok := m.(string)
+			if !ok {
+				return
+			}
+			if membership == "invite" {
+				logger := log.WithFields(log.Fields{
+					"room_id":         event.RoomID,
+					"service_user_id": config.UserID,
+					"inviter":         event.Sender,
+				})
+				logger.Print("Accepting invite from user")
+
+				if _, err := client.JoinRoom(event.RoomID, ""); err != nil {
+					logger.WithError(err).Print("Failed to join room")
+				} else {
+					logger.Print("Joined room")
+				}
+			}
+		})
+	}
+
 	go client.Sync()
 
 	return client, nil

src/github.com/matrix-org/go-neb/database/db.go

@@ -59,6 +59,15 @@ func (d *ServiceDB) StoreMatrixClientConfig(config types.ClientConfig) (oldConfi
 	return
 }
 
+// LoadMatrixClientConfigs loads all Matrix client configs from the database.
+func (d *ServiceDB) LoadMatrixClientConfigs() (configs []types.ClientConfig, err error) {
+	err = runTransaction(d.db, func(txn *sql.Tx) error {
+		configs, err = selectMatrixClientConfigsTxn(txn)
+		return err
+	})
+	return
+}
+
 // LoadMatrixClientConfig loads a Matrix client config from the database.
 // Returns sql.ErrNoRows if the client isn't in the database.
 func (d *ServiceDB) LoadMatrixClientConfig(userID string) (config types.ClientConfig, err error) {

src/github.com/matrix-org/go-neb/database/schema.go

@@ -64,6 +64,30 @@ func selectMatrixClientConfigTxn(txn *sql.Tx, userID string) (config types.Clien
 	return
 }
 
+const selectMatrixClientConfigsSQL = `
+SELECT client_json FROM matrix_clients
+`
+
+func selectMatrixClientConfigsTxn(txn *sql.Tx) (configs []types.ClientConfig, err error) {
+	rows, err := txn.Query(selectMatrixClientConfigsSQL)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var config types.ClientConfig
+		var configJSON []byte
+		if err = rows.Scan(&configJSON); err != nil {
+			return
+		}
+		if err = json.Unmarshal(configJSON, &config); err != nil {
+			return
+		}
+		configs = append(configs, config)
+	}
+	return
+}
+
 const insertMatrixClientConfigSQL = `
 INSERT INTO matrix_clients(
 	user_id, client_json, next_batch, time_added_ms, time_updated_ms

src/github.com/matrix-org/go-neb/goneb.go

@@ -51,6 +51,9 @@ func main() {
 	database.SetServiceDB(db)
 
 	clients := clients.New(db)
+	if err := clients.Start(); err != nil {
+		log.Panic(err)
+	}
 
 	http.Handle("/test", server.MakeJSONAPI(&heartbeatHandler{}))
 	http.Handle("/admin/getService", server.MakeJSONAPI(&getServiceHandler{db: db}))

src/github.com/matrix-org/go-neb/types/types.go

@@ -16,6 +16,8 @@ type ClientConfig struct {
 	UserID        string // The matrix UserId to connect with.
 	HomeserverURL string // A URL with the host and port of the matrix server. E.g. https://matrix.org:8448
 	AccessToken   string // The matrix access token to authenticate the requests with.
+	Sync          bool   // True to start a sync stream for this user
+	AutoJoinRooms bool   // True to automatically join all rooms for this user
 }
 
 // Check that the client has the correct fields.