Add message stats to reporting (#2748)

Since we're now listening on the `OutputRoomEvent` stream, we are able
to store messages stats.

9 files changed, 383 insertions, 11 deletions

diff --git a/userapi/consumers/roomserver.go b/userapi/consumers/roomserver.go
index a1287694..97c17e18 100644
--- a/userapi/consumers/roomserver.go
+++ b/userapi/consumers/roomserver.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/matrix-org/gomatrixserverlib"
@@ -23,6 +24,7 @@ import (
 	"github.com/matrix-org/dendrite/userapi/producers"
 	"github.com/matrix-org/dendrite/userapi/storage"
 	"github.com/matrix-org/dendrite/userapi/storage/tables"
+	userAPITypes "github.com/matrix-org/dendrite/userapi/types"
 	"github.com/matrix-org/dendrite/userapi/util"
 )
 
@@ -36,6 +38,11 @@ type OutputRoomEventConsumer struct {
 	topic        string
 	pgClient     pushgateway.Client
 	syncProducer *producers.SyncAPI
+	msgCounts    map[gomatrixserverlib.ServerName]userAPITypes.MessageStats
+	roomCounts   map[gomatrixserverlib.ServerName]map[string]bool // map from serverName to map from rommID to "isEncrypted"
+	lastUpdate   time.Time
+	countsLock   sync.Mutex
+	serverName   gomatrixserverlib.ServerName
 }
 
 func NewOutputRoomEventConsumer(
@@ -57,6 +64,11 @@ func NewOutputRoomEventConsumer(
 		pgClient:     pgClient,
 		rsAPI:        rsAPI,
 		syncProducer: syncProducer,
+		msgCounts:    map[gomatrixserverlib.ServerName]userAPITypes.MessageStats{},
+		roomCounts:   map[gomatrixserverlib.ServerName]map[string]bool{},
+		lastUpdate:   time.Now(),
+		countsLock:   sync.Mutex{},
+		serverName:   cfg.Matrix.ServerName,
 	}
 }
 
@@ -88,6 +100,10 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Ms
 		return true
 	}
 
+	if s.cfg.Matrix.ReportStats.Enabled {
+		go s.storeMessageStats(ctx, event.Type(), event.Sender(), event.RoomID())
+	}
+
 	log.WithFields(log.Fields{
 		"event_id":   event.EventID(),
 		"event_type": event.Type(),
@@ -107,6 +123,68 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Ms
 	return true
 }
 
+func (s *OutputRoomEventConsumer) storeMessageStats(ctx context.Context, eventType, eventSender, roomID string) {
+	s.countsLock.Lock()
+	defer s.countsLock.Unlock()
+
+	// reset the roomCounts on a day change
+	if s.lastUpdate.Day() != time.Now().Day() {
+		s.roomCounts[s.serverName] = make(map[string]bool)
+		s.lastUpdate = time.Now()
+	}
+
+	_, sender, err := gomatrixserverlib.SplitID('@', eventSender)
+	if err != nil {
+		return
+	}
+	msgCount := s.msgCounts[s.serverName]
+	roomCount := s.roomCounts[s.serverName]
+	if roomCount == nil {
+		roomCount = make(map[string]bool)
+	}
+	switch eventType {
+	case "m.room.message":
+		roomCount[roomID] = false
+		msgCount.Messages++
+		if sender == s.serverName {
+			msgCount.SentMessages++
+		}
+	case "m.room.encrypted":
+		roomCount[roomID] = true
+		msgCount.MessagesE2EE++
+		if sender == s.serverName {
+			msgCount.SentMessagesE2EE++
+		}
+	default:
+		return
+	}
+	s.msgCounts[s.serverName] = msgCount
+	s.roomCounts[s.serverName] = roomCount
+
+	for serverName, stats := range s.msgCounts {
+		var normalRooms, encryptedRooms int64 = 0, 0
+		for _, isEncrypted := range s.roomCounts[s.serverName] {
+			if isEncrypted {
+				encryptedRooms++
+			} else {
+				normalRooms++
+			}
+		}
+		err := s.db.UpsertDailyRoomsMessages(ctx, serverName, stats, normalRooms, encryptedRooms)
+		if err != nil {
+			log.WithError(err).Errorf("failed to upsert daily messages")
+		}
+		// Clear stats if we successfully stored it
+		if err == nil {
+			stats.Messages = 0
+			stats.SentMessages = 0
+			stats.MessagesE2EE = 0
+			stats.SentMessagesE2EE = 0
+			s.msgCounts[serverName] = stats
+		}
+	}
+}
+
 func (s *OutputRoomEventConsumer) processMessage(ctx context.Context, event *gomatrixserverlib.HeaderedEvent, streamPos uint64) error {
 	members, roomSize, err := s.localRoomMembers(ctx, event.RoomID())
 	if err != nil {
diff --git a/userapi/consumers/roomserver_test.go b/userapi/consumers/roomserver_test.go
index e4587670..265e3a3a 100644
--- a/userapi/consumers/roomserver_test.go
+++ b/userapi/consumers/roomserver_test.go
@@ -2,7 +2,10 @@ package consumers
 
 import (
 	"context"
+	"reflect"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/stretchr/testify/assert"
@@ -12,6 +15,7 @@ import (
 	"github.com/matrix-org/dendrite/test"
 	"github.com/matrix-org/dendrite/test/testrig"
 	"github.com/matrix-org/dendrite/userapi/storage"
+	userAPITypes "github.com/matrix-org/dendrite/userapi/types"
 )
 
 func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
@@ -132,3 +136,122 @@ func Test_evaluatePushRules(t *testing.T) {
 		}
 	})
 }
+
+func TestMessageStats(t *testing.T) {
+	type args struct {
+		eventType   string
+		eventSender string
+		roomID      string
+	}
+	tests := []struct {
+		name           string
+		args           args
+		ourServer      gomatrixserverlib.ServerName
+		lastUpdate     time.Time
+		initRoomCounts map[gomatrixserverlib.ServerName]map[string]bool
+		wantStats      userAPITypes.MessageStats
+	}{
+		{
+			name:      "m.room.create does not count as a message",
+			ourServer: "localhost",
+			args: args{
+				eventType:   "m.room.create",
+				eventSender: "@alice:localhost",
+			},
+		},
+		{
+			name:      "our server - message",
+			ourServer: "localhost",
+			args: args{
+				eventType:   "m.room.message",
+				eventSender: "@alice:localhost",
+				roomID:      "normalRoom",
+			},
+			wantStats: userAPITypes.MessageStats{Messages: 1, SentMessages: 1},
+		},
+		{
+			name:      "our server - E2EE message",
+			ourServer: "localhost",
+			args: args{
+				eventType:   "m.room.encrypted",
+				eventSender: "@alice:localhost",
+				roomID:      "encryptedRoom",
+			},
+			wantStats: userAPITypes.MessageStats{Messages: 1, SentMessages: 1, MessagesE2EE: 1, SentMessagesE2EE: 1},
+		},
+
+		{
+			name:      "remote server - message",
+			ourServer: "localhost",
+			args: args{
+				eventType:   "m.room.message",
+				eventSender: "@alice:remote",
+				roomID:      "normalRoom",
+			},
+			wantStats: userAPITypes.MessageStats{Messages: 2, SentMessages: 1, MessagesE2EE: 1, SentMessagesE2EE: 1},
+		},
+		{
+			name:      "remote server - E2EE message",
+			ourServer: "localhost",
+			args: args{
+				eventType:   "m.room.encrypted",
+				eventSender: "@alice:remote",
+				roomID:      "encryptedRoom",
+			},
+			wantStats: userAPITypes.MessageStats{Messages: 2, SentMessages: 1, MessagesE2EE: 2, SentMessagesE2EE: 1},
+		},
+		{
+			name:       "day change creates a new room map",
+			ourServer:  "localhost",
+			lastUpdate: time.Now().Add(-time.Hour * 24),
+			initRoomCounts: map[gomatrixserverlib.ServerName]map[string]bool{
+				"localhost": {"encryptedRoom": true},
+			},
+			args: args{
+				eventType:   "m.room.encrypted",
+				eventSender: "@alice:remote",
+				roomID:      "someOtherRoom",
+			},
+			wantStats: userAPITypes.MessageStats{Messages: 2, SentMessages: 1, MessagesE2EE: 3, SentMessagesE2EE: 1},
+		},
+	}
+
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		db, close := mustCreateDatabase(t, dbType)
+		defer close()
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				if tt.lastUpdate.IsZero() {
+					tt.lastUpdate = time.Now()
+				}
+				if tt.initRoomCounts == nil {
+					tt.initRoomCounts = map[gomatrixserverlib.ServerName]map[string]bool{}
+				}
+				s := &OutputRoomEventConsumer{
+					db:         db,
+					msgCounts:  map[gomatrixserverlib.ServerName]userAPITypes.MessageStats{},
+					roomCounts: tt.initRoomCounts,
+					countsLock: sync.Mutex{},
+					lastUpdate: tt.lastUpdate,
+					serverName: tt.ourServer,
+				}
+				s.storeMessageStats(context.Background(), tt.args.eventType, tt.args.eventSender, tt.args.roomID)
+				t.Logf("%+v", s.roomCounts)
+				gotStats, activeRooms, activeE2EERooms, err := db.DailyRoomsMessages(context.Background(), tt.ourServer)
+				if err != nil {
+					t.Fatalf("unexpected error: %s", err)
+				}
+				if !reflect.DeepEqual(gotStats, tt.wantStats) {
+					t.Fatalf("expected %+v, got %+v", tt.wantStats, gotStats)
+				}
+				if tt.args.eventType == "m.room.encrypted" && activeE2EERooms != 1 {
+					t.Fatalf("expected room to be activeE2EE")
+				}
+				if tt.args.eventType == "m.room.message" && activeRooms != 1 {
+					t.Fatalf("expected room to be active")
+				}
+			})
+		}
+	})
+}
diff --git a/userapi/storage/interface.go b/userapi/storage/interface.go
index fb12b53a..28ef2655 100644
--- a/userapi/storage/interface.go
+++ b/userapi/storage/interface.go
@@ -19,6 +19,8 @@ import (
 	"encoding/json"
 	"errors"
 
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
 	"github.com/matrix-org/dendrite/internal/pushrules"
 	"github.com/matrix-org/dendrite/userapi/api"
@@ -144,6 +146,8 @@ type Database interface {
 
 type Statistics interface {
 	UserStatistics(ctx context.Context) (*types.UserStatistics, *types.DatabaseEngine, error)
+	DailyRoomsMessages(ctx context.Context, serverName gomatrixserverlib.ServerName) (stats types.MessageStats, activeRooms, activeE2EERooms int64, err error)
+	UpsertDailyRoomsMessages(ctx context.Context, serverName gomatrixserverlib.ServerName, stats types.MessageStats, activeRooms, activeE2EERooms int64) error
 }
 
 // Err3PIDInUse is the error returned when trying to save an association involving
diff --git a/userapi/storage/postgres/stats_table.go b/userapi/storage/postgres/stats_table.go
index 20eb0bf4..f62467fa 100644
--- a/userapi/storage/postgres/stats_table.go
+++ b/userapi/storage/postgres/stats_table.go
@@ -20,13 +20,14 @@ import (
 	"time"
 
 	"github.com/lib/pq"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/dendrite/userapi/storage/tables"
 	"github.com/matrix-org/dendrite/userapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/sirupsen/logrus"
 )
 
 const userDailyVisitsSchema = `
@@ -43,6 +44,35 @@ CREATE INDEX IF NOT EXISTS userapi_daily_visits_timestamp_idx ON userapi_daily_v
 CREATE INDEX IF NOT EXISTS userapi_daily_visits_localpart_timestamp_idx ON userapi_daily_visits(localpart, timestamp);
 `
 
+const messagesDailySchema = `
+CREATE TABLE IF NOT EXISTS userapi_daily_stats (
+	timestamp BIGINT NOT NULL,
+	server_name TEXT NOT NULL,
+	messages BIGINT NOT NULL,
+	sent_messages BIGINT NOT NULL,
+	e2ee_messages BIGINT NOT NULL,
+	sent_e2ee_messages BIGINT NOT NULL,
+	active_rooms BIGINT NOT NULL,
+	active_e2ee_rooms BIGINT NOT NULL,
+	CONSTRAINT daily_stats_unique UNIQUE (timestamp, server_name)
+);
+`
+
+const upsertDailyMessagesSQL = `
+	INSERT INTO userapi_daily_stats AS u (timestamp, server_name, messages, sent_messages, e2ee_messages, sent_e2ee_messages, active_rooms, active_e2ee_rooms)
+	VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT ON CONSTRAINT daily_stats_unique
+	DO UPDATE SET
+	    messages=u.messages+excluded.messages, sent_messages=u.sent_messages+excluded.sent_messages,
+	    e2ee_messages=u.e2ee_messages+excluded.e2ee_messages, sent_e2ee_messages=u.sent_e2ee_messages+excluded.sent_e2ee_messages,
+		active_rooms=GREATEST($7, u.active_rooms), active_e2ee_rooms=GREATEST($8, u.active_e2ee_rooms)
+`
+
+const selectDailyMessagesSQL = `
+	SELECT messages, sent_messages, e2ee_messages, sent_e2ee_messages, active_rooms, active_e2ee_rooms
+	FROM userapi_daily_stats
+	WHERE server_name = $1 AND timestamp = $2;
+`
+
 const countUsersLastSeenAfterSQL = "" +
 	"SELECT COUNT(*) FROM (" +
 	" SELECT localpart FROM userapi_devices WHERE last_seen_ts > $1 " +
@@ -170,6 +200,8 @@ type statsStatements struct {
 	countUserByAccountTypeStmt    *sql.Stmt
 	countRegisteredUserByTypeStmt *sql.Stmt
 	dbEngineVersionStmt           *sql.Stmt
+	upsertMessagesStmt            *sql.Stmt
+	selectDailyMessagesStmt       *sql.Stmt
 }
 
 func NewPostgresStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName) (tables.StatsTable, error) {
@@ -182,6 +214,10 @@ func NewPostgresStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName)
 	if err != nil {
 		return nil, err
 	}
+	_, err = db.Exec(messagesDailySchema)
+	if err != nil {
+		return nil, err
+	}
 	go s.startTimers()
 	return s, sqlutil.StatementList{
 		{&s.countUsersLastSeenAfterStmt, countUsersLastSeenAfterSQL},
@@ -191,6 +227,8 @@ func NewPostgresStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName)
 		{&s.countUserByAccountTypeStmt, countUserByAccountTypeSQL},
 		{&s.countRegisteredUserByTypeStmt, countRegisteredUserByTypeStmt},
 		{&s.dbEngineVersionStmt, queryDBEngineVersion},
+		{&s.upsertMessagesStmt, upsertDailyMessagesSQL},
+		{&s.selectDailyMessagesStmt, selectDailyMessagesSQL},
 	}.Prepare(db)
 }
 
@@ -435,3 +473,34 @@ func (s *statsStatements) UpdateUserDailyVisits(
 	}
 	return err
 }
+
+func (s *statsStatements) UpsertDailyStats(
+	ctx context.Context, txn *sql.Tx,
+	serverName gomatrixserverlib.ServerName, stats types.MessageStats,
+	activeRooms, activeE2EERooms int64,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.upsertMessagesStmt)
+	timestamp := time.Now().Truncate(time.Hour * 24)
+	_, err := stmt.ExecContext(ctx,
+		gomatrixserverlib.AsTimestamp(timestamp),
+		serverName,
+		stats.Messages, stats.SentMessages, stats.MessagesE2EE, stats.SentMessagesE2EE,
+		activeRooms, activeE2EERooms,
+	)
+	return err
+}
+
+func (s *statsStatements) DailyRoomsMessages(
+	ctx context.Context, txn *sql.Tx,
+	serverName gomatrixserverlib.ServerName,
+) (msgStats types.MessageStats, activeRooms, activeE2EERooms int64, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectDailyMessagesStmt)
+	timestamp := time.Now().Truncate(time.Hour * 24)
+
+	err = stmt.QueryRowContext(ctx, serverName, gomatrixserverlib.AsTimestamp(timestamp)).
+		Scan(&msgStats.Messages, &msgStats.SentMessages, &msgStats.MessagesE2EE, &msgStats.SentMessagesE2EE, &activeRooms, &activeE2EERooms)
+	if err != nil && err != sql.ErrNoRows {
+		return msgStats, 0, 0, err
+	}
+	return msgStats, activeRooms, activeE2EERooms, nil
+}
diff --git a/userapi/storage/shared/storage.go b/userapi/storage/shared/storage.go
index f8b6ad31..f8b8d02c 100644
--- a/userapi/storage/shared/storage.go
+++ b/userapi/storage/shared/storage.go
@@ -29,13 +29,12 @@ import (
 	"github.com/matrix-org/gomatrixserverlib"
 	"golang.org/x/crypto/bcrypt"
 
-	"github.com/matrix-org/dendrite/userapi/types"
-
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
 	"github.com/matrix-org/dendrite/internal/pushrules"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/dendrite/userapi/storage/tables"
+	"github.com/matrix-org/dendrite/userapi/types"
 )
 
 // Database represents an account database
@@ -808,3 +807,15 @@ func (d *Database) RemovePushers(
 func (d *Database) UserStatistics(ctx context.Context) (*types.UserStatistics, *types.DatabaseEngine, error) {
 	return d.Stats.UserStatistics(ctx, nil)
 }
+
+func (d *Database) UpsertDailyRoomsMessages(ctx context.Context, serverName gomatrixserverlib.ServerName, stats types.MessageStats, activeRooms, activeE2EERooms int64) error {
+	return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		return d.Stats.UpsertDailyStats(ctx, txn, serverName, stats, activeRooms, activeE2EERooms)
+	})
+}
+
+func (d *Database) DailyRoomsMessages(
+	ctx context.Context, serverName gomatrixserverlib.ServerName,
+) (stats types.MessageStats, activeRooms, activeE2EERooms int64, err error) {
+	return d.Stats.DailyRoomsMessages(ctx, nil, serverName)
+}
diff --git a/userapi/storage/sqlite3/stats_table.go b/userapi/storage/sqlite3/stats_table.go
index 35e3c653..a1365c94 100644
--- a/userapi/storage/sqlite3/stats_table.go
+++ b/userapi/storage/sqlite3/stats_table.go
@@ -44,6 +44,35 @@ CREATE INDEX IF NOT EXISTS userapi_daily_visits_timestamp_idx ON userapi_daily_v
 CREATE INDEX IF NOT EXISTS userapi_daily_visits_localpart_timestamp_idx ON userapi_daily_visits(localpart, timestamp);
 `
 
+const messagesDailySchema = `
+CREATE TABLE IF NOT EXISTS userapi_daily_stats (
+	timestamp BIGINT NOT NULL,
+	server_name TEXT NOT NULL,
+	messages BIGINT NOT NULL,
+	sent_messages BIGINT NOT NULL,
+	e2ee_messages BIGINT NOT NULL,
+	sent_e2ee_messages BIGINT NOT NULL,
+	active_rooms BIGINT NOT NULL,
+	active_e2ee_rooms BIGINT NOT NULL,
+	CONSTRAINT daily_stats_unique UNIQUE (timestamp, server_name)
+);
+`
+
+const upsertDailyMessagesSQL = `
+	INSERT INTO userapi_daily_stats (timestamp, server_name, messages, sent_messages, e2ee_messages, sent_e2ee_messages, active_rooms, active_e2ee_rooms)
+	VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (timestamp, server_name)
+	DO UPDATE SET
+	    messages=messages+excluded.messages, sent_messages=sent_messages+excluded.sent_messages,
+	    e2ee_messages=e2ee_messages+excluded.e2ee_messages, sent_e2ee_messages=sent_e2ee_messages+excluded.sent_e2ee_messages,
+	    active_rooms=MAX($7, active_rooms), active_e2ee_rooms=MAX($8, active_e2ee_rooms)
+`
+
+const selectDailyMessagesSQL = `
+	SELECT messages, sent_messages, e2ee_messages, sent_e2ee_messages, active_rooms, active_e2ee_rooms
+	FROM userapi_daily_stats
+	WHERE server_name = $1 AND timestamp = $2;
+`
+
 const countUsersLastSeenAfterSQL = "" +
 	"SELECT COUNT(*) FROM (" +
 	" SELECT localpart FROM userapi_devices WHERE last_seen_ts > $1 " +
@@ -176,6 +205,8 @@ type statsStatements struct {
 	countUserByAccountTypeStmt    *sql.Stmt
 	countRegisteredUserByTypeStmt *sql.Stmt
 	dbEngineVersionStmt           *sql.Stmt
+	upsertMessagesStmt            *sql.Stmt
+	selectDailyMessagesStmt       *sql.Stmt
 }
 
 func NewSQLiteStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName) (tables.StatsTable, error) {
@@ -189,6 +220,10 @@ func NewSQLiteStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName) (t
 	if err != nil {
 		return nil, err
 	}
+	_, err = db.Exec(messagesDailySchema)
+	if err != nil {
+		return nil, err
+	}
 	go s.startTimers()
 	return s, sqlutil.StatementList{
 		{&s.countUsersLastSeenAfterStmt, countUsersLastSeenAfterSQL},
@@ -198,6 +233,8 @@ func NewSQLiteStatsTable(db *sql.DB, serverName gomatrixserverlib.ServerName) (t
 		{&s.countUserByAccountTypeStmt, countUserByAccountTypeSQL},
 		{&s.countRegisteredUserByTypeStmt, countRegisteredUserByTypeSQL},
 		{&s.dbEngineVersionStmt, queryDBEngineVersion},
+		{&s.upsertMessagesStmt, upsertDailyMessagesSQL},
+		{&s.selectDailyMessagesStmt, selectDailyMessagesSQL},
 	}.Prepare(db)
 }
 
@@ -451,3 +488,34 @@ func (s *statsStatements) UpdateUserDailyVisits(
 	}
 	return err
 }
+
+func (s *statsStatements) UpsertDailyStats(
+	ctx context.Context, txn *sql.Tx,
+	serverName gomatrixserverlib.ServerName, stats types.MessageStats,
+	activeRooms, activeE2EERooms int64,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.upsertMessagesStmt)
+	timestamp := time.Now().Truncate(time.Hour * 24)
+	_, err := stmt.ExecContext(ctx,
+		gomatrixserverlib.AsTimestamp(timestamp),
+		serverName,
+		stats.Messages, stats.SentMessages, stats.MessagesE2EE, stats.SentMessagesE2EE,
+		activeRooms, activeE2EERooms,
+	)
+	return err
+}
+
+func (s *statsStatements) DailyRoomsMessages(
+	ctx context.Context, txn *sql.Tx,
+	serverName gomatrixserverlib.ServerName,
+) (msgStats types.MessageStats, activeRooms, activeE2EERooms int64, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectDailyMessagesStmt)
+	timestamp := time.Now().Truncate(time.Hour * 24)
+
+	err = stmt.QueryRowContext(ctx, serverName, gomatrixserverlib.AsTimestamp(timestamp)).
+		Scan(&msgStats.Messages, &msgStats.SentMessages, &msgStats.MessagesE2EE, &msgStats.SentMessagesE2EE, &activeRooms, &activeE2EERooms)
+	if err != nil && err != sql.ErrNoRows {
+		return msgStats, 0, 0, err
+	}
+	return msgStats, activeRooms, activeE2EERooms, nil
+}
diff --git a/userapi/storage/tables/interface.go b/userapi/storage/tables/interface.go
index 1b239e44..5e1dd097 100644
--- a/userapi/storage/tables/interface.go
+++ b/userapi/storage/tables/interface.go
@@ -20,6 +20,8 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
 	"github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/dendrite/userapi/types"
@@ -115,7 +117,9 @@ type NotificationTable interface {
 
 type StatsTable interface {
 	UserStatistics(ctx context.Context, txn *sql.Tx) (*types.UserStatistics, *types.DatabaseEngine, error)
+	DailyRoomsMessages(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) (msgStats types.MessageStats, activeRooms, activeE2EERooms int64, err error)
 	UpdateUserDailyVisits(ctx context.Context, txn *sql.Tx, startTime, lastUpdate time.Time) error
+	UpsertDailyStats(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, stats types.MessageStats, activeRooms, activeE2EERooms int64) error
 }
 
 type NotificationFilter uint32
diff --git a/userapi/types/statistics.go b/userapi/types/statistics.go
index 09564f78..b74e32ad 100644
--- a/userapi/types/statistics.go
+++ b/userapi/types/statistics.go
@@ -28,3 +28,10 @@ type DatabaseEngine struct {
 	Engine  string
 	Version string
 }
+
+type MessageStats struct {
+	Messages         int64
+	SentMessages     int64
+	MessagesE2EE     int64
+	SentMessagesE2EE int64
+}
diff --git a/userapi/util/phonehomestats.go b/userapi/util/phonehomestats.go
index b17f6206..6f36568c 100644
--- a/userapi/util/phonehomestats.go
+++ b/userapi/util/phonehomestats.go
@@ -24,11 +24,12 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/userapi/storage"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/sirupsen/logrus"
 )
 
 type phoneHomeStats struct {
@@ -109,12 +110,19 @@ func (p *phoneHomeStats) collect() {
 	}
 
 	// message and room stats
-	// TODO: Find a solution to actually set these values
+	// TODO: Find a solution to actually set this value
 	p.stats["total_room_count"] = 0
-	p.stats["daily_messages"] = 0
-	p.stats["daily_sent_messages"] = 0
-	p.stats["daily_e2ee_messages"] = 0
-	p.stats["daily_sent_e2ee_messages"] = 0
+
+	messageStats, activeRooms, activeE2EERooms, err := p.db.DailyRoomsMessages(ctx, p.serverName)
+	if err != nil {
+		logrus.WithError(err).Warn("unable to query message stats, using default values")
+	}
+	p.stats["daily_messages"] = messageStats.Messages
+	p.stats["daily_sent_messages"] = messageStats.SentMessages
+	p.stats["daily_e2ee_messages"] = messageStats.MessagesE2EE
+	p.stats["daily_sent_e2ee_messages"] = messageStats.SentMessagesE2EE
+	p.stats["daily_active_rooms"] = activeRooms
+	p.stats["daily_active_e2ee_rooms"] = activeE2EERooms
 
 	// user stats and DB engine
 	userStats, db, err := p.db.UserStatistics(ctx)