status-go/protocol/anonmetrics/server.go

190 lines
4.0 KiB
Go

package anonmetrics
import (
"database/sql"
// Import postgres driver
_ "github.com/lib/pq"
"go.uber.org/zap"
"github.com/status-im/migrate/v4"
"github.com/status-im/migrate/v4/database/postgres"
bindata "github.com/status-im/migrate/v4/source/go_bindata"
"github.com/status-im/status-go/appmetrics"
"github.com/status-im/status-go/protocol/anonmetrics/migrations"
"github.com/status-im/status-go/protocol/protobuf"
)
const ActiveServerPhrase = "I was thinking that it would be a pretty nice idea if the server functionality was working now, I express gratitude in the anticipation"
type ServerConfig struct {
Enabled bool
PostgresURI string
Active string
}
type Server struct {
Config *ServerConfig
Logger *zap.Logger
PostgresDB *sql.DB
}
func NewServer(postgresURI string) (*Server, error) {
postgresMigration := bindata.Resource(migrations.AssetNames(), migrations.Asset)
db, err := NewMigratedDB(postgresURI, postgresMigration)
if err != nil {
return nil, err
}
return &Server{
PostgresDB: db,
}, nil
}
func (s *Server) Stop() error {
if s.PostgresDB != nil {
return s.PostgresDB.Close()
}
return nil
}
func (s *Server) StoreMetrics(appMetricsBatch *protobuf.AnonymousMetricBatch) (appMetrics []*appmetrics.AppMetric, err error) {
if s.Config.Active != ActiveServerPhrase {
return nil, nil
}
s.Logger.Debug("StoreMetrics() triggered with payload",
zap.Reflect("appMetricsBatch", appMetricsBatch))
appMetrics, err = adaptProtoBatchToModels(appMetricsBatch)
if err != nil {
return
}
var (
tx *sql.Tx
insert *sql.Stmt
)
// start txn
tx, err = s.PostgresDB.Begin()
if err != nil {
return
}
defer func() {
if err == nil {
err = tx.Commit()
return
}
_ = tx.Rollback()
}()
//noinspection ALL
query := `INSERT INTO app_metrics (message_id, event, value, app_version, operating_system, session_id, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (message_id) DO NOTHING;`
insert, err = tx.Prepare(query)
if err != nil {
return
}
for _, metric := range appMetrics {
_, err = insert.Exec(
metric.MessageID,
metric.Event,
metric.Value,
metric.AppVersion,
metric.OS,
metric.SessionID,
metric.CreatedAt,
)
if err != nil {
return
}
}
return
}
func (s *Server) getFromRows(rows *sql.Rows) (appMetrics []appmetrics.AppMetric, err error) {
for rows.Next() {
metric := appmetrics.AppMetric{}
err = rows.Scan(
&metric.ID,
&metric.MessageID,
&metric.Event,
&metric.Value,
&metric.AppVersion,
&metric.OS,
&metric.SessionID,
&metric.CreatedAt,
&metric.Processed,
&metric.ReceivedAt,
)
if err != nil {
return nil, err
}
appMetrics = append(appMetrics, metric)
}
return appMetrics, nil
}
func (s *Server) GetAppMetrics(limit int, offset int) ([]appmetrics.AppMetric, error) {
if s.Config.Active != ActiveServerPhrase {
return nil, nil
}
rows, err := s.PostgresDB.Query("SELECT id, message_id, event, value, app_version, operating_system, session_id, created_at, processed, received_at FROM app_metrics LIMIT $1 OFFSET $2", limit, offset)
if err != nil {
return nil, err
}
defer rows.Close()
return s.getFromRows(rows)
}
func NewMigratedDB(uri string, migrationResource *bindata.AssetSource) (*sql.DB, error) {
db, err := sql.Open("postgres", uri)
if err != nil {
return nil, err
}
if err := setup(db, migrationResource); err != nil {
return nil, err
}
return db, nil
}
func setup(d *sql.DB, migrationResource *bindata.AssetSource) error {
m, err := MakeMigration(d, migrationResource)
if err != nil {
return err
}
if err = m.Up(); err != migrate.ErrNoChange {
return err
}
return nil
}
func MakeMigration(d *sql.DB, migrationResource *bindata.AssetSource) (*migrate.Migrate, error) {
source, err := bindata.WithInstance(migrationResource)
if err != nil {
return nil, err
}
driver, err := postgres.WithInstance(d, &postgres.Config{})
if err != nil {
return nil, err
}
return migrate.NewWithInstance(
"go-bindata",
source,
"postgres",
driver)
}