feat: pg as storage: contacts

.golangci.yml

@@ -50,6 +50,8 @@ linters:
     - wrapcheck
     - zerologlint
     - modernize
+    - goconst
+    - prealloc
   settings:
     gocyclo:
       min-complexity: 15

cmd/api/config.go

@@ -15,6 +15,7 @@ import (
 
 type config struct {
 	Redis               cmd.RedisConfig               `yaml:"redis"`
+	Database            cmd.DatabaseConfig            `yaml:"database"`
 	Logger              cmd.LoggerConfig              `yaml:"log"`
 	API                 apiConfig                     `yaml:"api"`
 	Web                 webConfig                     `yaml:"web"`

cmd/cli/notification_history.go

@@ -115,6 +115,7 @@ func mergeNotificationHistory(logger moira.Logger, database moira.Database) erro
 						Member: eventBytes,
 					})
 				}
+
 				return nil
 			})
 			if err != nil {
@@ -128,13 +129,15 @@ func mergeNotificationHistory(logger moira.Logger, database moira.Database) erro
 				for _, id := range contactIDs {
 					pipe.Del(connector.Context(), id)
 				}
+
 				return nil
 			})
 			if err != nil {
 				return fmt.Errorf("failed to delete previous notification history: %w", err)
 			}
 
 			var totalDelCount int64
+
 			for i, cmd := range cmds {
 				deleted, err := cmd.(*goredis.IntCmd).Result()
 				if err != nil {
@@ -143,6 +146,7 @@ func mergeNotificationHistory(logger moira.Logger, database moira.Database) erro
 						Error(err).
 						Msg("failed to delete")
 				}
+
 				totalDelCount += deleted
 			}
 

cmd/config.go

@@ -114,6 +114,20 @@ func (config *RedisConfig) GetSettings() redis.DatabaseConfig {
 	}
 }
 
+type PostgresqlConfig struct {
+	Master struct {
+		ConnectionString string
+	}
+	Replicas []struct {
+		ConnectionString string
+	}
+}
+
+type DatabaseConfig struct {
+	Redis      RedisConfig
+	Postgresql PostgresqlConfig
+}
+
 // NotificationHistoryConfig is the config which coordinates interaction with notification statistics.
 // E.g. how much time should we store it, or how many history items can we request from database.
 type NotificationHistoryConfig struct {

database/postgresql/config.go

@@ -0,0 +1,10 @@
+package postgresql
+
+type DatabaseConfig struct {
+	Master   ReplicaConfig
+	Replicas []ReplicaConfig
+}
+
+type ReplicaConfig struct {
+	ConnectionString string
+}

database/postgresql/contact.go

@@ -0,0 +1,311 @@
+package postgresql
+
+import (
+	"fmt"
+
+	"github.com/lib/pq"
+	"github.com/moira-alert/moira"
+)
+
+func (connector *DbConnector) GetContact(contactId string) (moira.ContactData, error) {
+	query := `
+SELECT
+    c.type,
+    c.name,
+    c.value,
+    c.extra_message,
+
+    u.login,
+    t.team_id
+
+FROM contacts c
+LEFT JOIN users u
+    ON c.user_id = u.id
+LEFT JOIN teams t
+    ON c.team_id = t.id
+WHERE c.contact_id = $1 LIMIT 1;
+	`
+	requester := func(db RDB) (moira.ContactData, error) {
+		row := db.QueryRowContext(connector.ctx, query, contactId)
+		var (
+			contactType string
+			name string
+			value string
+			extraMessage string
+			userLogin *string
+			teamId *string
+		)
+		err := row.Scan(&contactType, &name, &value, &extraMessage, &userLogin, &teamId)
+		return moira.ContactData{
+			Type: contactType,
+			Name: name,
+			Value: value,
+			ID: contactId,
+			User: moira.UseString(userLogin),
+			Team: moira.UseString(teamId),
+			ExtraMessage: extraMessage,
+		}, err
+	}
+
+	respFromReplica, err := requester(connector.db.Replica())
+	if err == nil {
+		return respFromReplica, nil
+	}
+
+	respFromMaster, err := requester(connector.db.Master())
+	return respFromMaster, err
+}
+
+func (connector *DbConnector) GetContacts(contactIDs []string) ([]*moira.ContactData, error) {
+	query := `
+SELECT
+		c.contact_id,
+    c.type,
+    c.name,
+    c.value,
+    c.extra_message,
+
+    u.login,
+    t.team_id
+
+FROM contacts c
+LEFT JOIN users u
+    ON c.user_id = u.id
+LEFT JOIN teams t
+    ON c.team_id = t.id
+WHERE c.contact_id = ANY($1);
+	`
+
+	responseFromReplica, err := connector.db.Replica().QueryContext(connector.ctx, query, pq.Array(contactIDs))
+	if err != nil {
+		return nil, fmt.Errorf("get contacts by ids error: %w", err)
+	}
+
+	contacts := make([]*moira.ContactData, 0)
+	for responseFromReplica.Next() {
+		var (
+			contactId string
+			contactType string
+			name string
+			value string
+			extraMessage string
+			userLogin *string
+			teamId *string
+		)
+		err := responseFromReplica.Scan(&contactId, &contactType, &name, &value, &extraMessage, &userLogin, &teamId)
+		if err != nil {
+			return nil, fmt.Errorf("unmarshaling error on get contacts: %w", err)
+		}
+		contacts = append(contacts, &moira.ContactData{
+			Type: contactType,
+			Name: name,
+			Value: value,
+			ID: contactId,
+			User: moira.UseString(userLogin),
+			Team: moira.UseString(teamId),
+			ExtraMessage: extraMessage,
+		})
+	}
+	return contacts, nil
+}
+
+func (connector *DbConnector) GetAllContacts() ([]*moira.ContactData, error) {
+	query := `
+SELECT
+		c.contact_id,
+    c.type,
+    c.name,
+    c.value,
+    c.extra_message,
+
+    u.login,
+    t.team_id
+
+FROM contacts c
+LEFT JOIN users u
+    ON c.user_id = u.id
+LEFT JOIN teams t
+    ON c.team_id = t.id;
+	`
+
+	responseFromReplica, err := connector.db.Replica().QueryContext(connector.ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("get contacts by ids error: %w", err)
+	}
+
+	contacts := make([]*moira.ContactData, 0)
+	for responseFromReplica.Next() {
+		var (
+			contactId string
+			contactType string
+			name string
+			value string
+			extraMessage string
+			userLogin *string
+			teamId *string
+		)
+		err := responseFromReplica.Scan(&contactId, &contactType, &name, &value, &extraMessage, &userLogin, &teamId)
+		if err != nil {
+			return nil, fmt.Errorf("unmarshaling error on get all contacts: %w", err)
+		}
+		contacts = append(contacts, &moira.ContactData{
+			Type: contactType,
+			Name: name,
+			Value: value,
+			ID: contactId,
+			User: moira.UseString(userLogin),
+			Team: moira.UseString(teamId),
+			ExtraMessage: extraMessage,
+		})
+	}
+	return contacts, nil
+}
+
+func (connector *DbConnector) SaveContact(contact *moira.ContactData) error {
+	if contact == nil {
+		return nil
+	}
+
+	queryWithUser := `
+WITH selected_user AS (
+    SELECT id
+    FROM users
+    WHERE login = $1
+)
+INSERT INTO contacts (
+    type,
+    name,
+    value,
+    contact_id,
+    user_id,
+    team_id,
+    extra_message
+)
+SELECT
+    $2,
+    $3,
+    $4,
+    $5,
+    selected_user.id,
+    NULL,
+    $6
+FROM selected_user;
+	`
+	queryWithTeam := `
+INSERT INTO contacts (
+    type,
+    name,
+    value,
+    contact_id,
+    user_id,
+    team_id,
+    extra_message
+)
+SELECT
+    $2,
+    $3,
+    $4,
+    $5,
+    NULL,
+    t.id,
+    $6
+FROM teams t
+WHERE t.team_id = $1;
+	`
+
+	query := queryWithUser
+	ownerId := contact.User
+	if contact.Team != "" {
+		query = queryWithTeam
+		ownerId = contact.Team
+	}
+	_, err := connector.db.Master().ExecContext(connector.ctx, query,
+		ownerId,
+		contact.Type,
+		contact.Name,
+		contact.Value,
+		contact.ID,
+		contact.ExtraMessage,
+	)
+	if err != nil {
+		return fmt.Errorf("save contact error: %w", err)
+	}
+	return nil
+}
+
+func (connector *DbConnector) RemoveContact(contactID string) error {
+	//TODO: maybe we should use removing by setting flag instead of hard delete
+	query := `
+DELETE FROM contacts WHERE contact_id = $1
+	`
+	_, err := connector.db.Master().ExecContext(connector.ctx, query, contactID)
+	if err != nil {
+		return fmt.Errorf("remove contact error: %w", err)
+	}
+	return nil
+}
+
+func (connector *DbConnector) GetUserContactIDs(login string) ([]string, error) {
+	query := `
+WITH selected_user AS (
+    SELECT id
+    FROM users
+    WHERE login = $1
+		LIMIT 1
+)
+SELECT contact_id
+FROM contacts
+WHERE user_id = (
+	SELECT id
+	FROM selected_user
+)
+	`
+	responseFromReplica, err := connector.db.Replica().QueryContext(connector.ctx, query, login)
+	if err != nil {
+		return nil, fmt.Errorf("get contacts by user login error: %w", err)
+	}
+
+	contacts := make([]string, 0)
+	for responseFromReplica.Next() {
+		var contactId string
+		err := responseFromReplica.Scan(&contactId)
+		if err != nil {
+			return nil, fmt.Errorf("unmarshaling error on get contacts ids: %w", err)
+		}
+		contacts = append(contacts, contactId)
+	}
+	return contacts, nil
+}
+
+func (connector *DbConnector) GetTeamContactIDs(teamId string) ([]string, error) {
+	query := `
+WITH selected_team AS (
+    SELECT id
+    FROM teams
+    WHERE team_id = $1
+		LIMIT 1
+)
+SELECT contact_id
+FROM contacts
+WHERE team_id = (
+	SELECT id
+	FROM selected_team
+)
+	`
+	responseFromReplica, err := connector.db.Replica().QueryContext(connector.ctx, query, teamId)
+	if err != nil {
+		return nil, fmt.Errorf("get contacts by team id error: %w", err)
+	}
+
+	contacts := make([]string, 0)
+	for responseFromReplica.Next() {
+		var contactId string
+		err := responseFromReplica.Scan(&contactId)
+		if err != nil {
+			return nil, fmt.Errorf("unmarshaling error on get contacts ids: %w", err)
+		}
+		contacts = append(contacts, contactId)
+	}
+	return contacts, nil
+}
+