status-go/protocol/communities/persistence.go

346 lines
11 KiB
Go
Raw Normal View History

package communities
import (
2021-01-11 10:32:51 +00:00
"context"
"crypto/ecdsa"
"database/sql"
2021-01-11 10:32:51 +00:00
"errors"
"github.com/golang/protobuf/proto"
"go.uber.org/zap"
"github.com/status-im/status-go/eth-node/crypto"
2021-01-11 10:32:51 +00:00
"github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/protobuf"
)
type Persistence struct {
db *sql.DB
logger *zap.Logger
}
const communitiesBaseQuery = `SELECT c.id, c.private_key, c.description,c.joined,c.verified,c.muted,r.clock FROM communities_communities c LEFT JOIN communities_requests_to_join r ON c.id = r.community_id AND r.public_key = ?`
2021-01-11 10:32:51 +00:00
func (p *Persistence) SaveCommunity(community *Community) error {
id := community.ID()
privateKey := community.PrivateKey()
description, err := community.ToBytes()
if err != nil {
return err
}
_, err = p.db.Exec(`INSERT INTO communities_communities (id, private_key, description, joined, verified) VALUES (?, ?, ?, ?, ?)`, id, crypto.FromECDSA(privateKey), description, community.config.Joined, community.config.Verified)
return err
}
func (p *Persistence) ShouldHandleSyncCommunity(community *protobuf.SyncCommunity) (bool, error) {
// TODO see if there is a way to make this more elegant
// Keep the "*".
// When the test for this function fails because the table has changed we should update sync functionality
qr := p.db.QueryRow(`SELECT * FROM communities_communities WHERE id = ? AND synced_at > ?`, community.Id, community.Clock)
_, err := p.scanRowToStruct(qr.Scan)
switch err {
case sql.ErrNoRows:
// Query does not match, therefore synced_at value is not older than the new clock value or id was not found
return true, nil
case nil:
// Error is nil, therefore query matched and synced_at is older than the new clock
return false, nil
default:
// Error is not nil and is not sql.ErrNoRows, therefore pass out the error
return false, err
}
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) queryCommunities(memberIdentity *ecdsa.PublicKey, query string) (response []*Community, err error) {
2021-01-11 10:32:51 +00:00
rows, err := p.db.Query(query, common.PubkeyToHex(memberIdentity))
if err != nil {
return nil, err
}
2020-12-22 11:00:13 +00:00
defer func() {
if err != nil {
// Don't shadow original error
_ = rows.Close()
return
}
err = rows.Close()
}()
for rows.Next() {
var publicKeyBytes, privateKeyBytes, descriptionBytes []byte
var joined, verified, muted bool
2021-01-11 10:32:51 +00:00
var requestedToJoinAt sql.NullInt64
err := rows.Scan(&publicKeyBytes, &privateKeyBytes, &descriptionBytes, &joined, &verified, &muted, &requestedToJoinAt)
if err != nil {
return nil, err
}
org, err := unmarshalCommunityFromDB(memberIdentity, publicKeyBytes, privateKeyBytes, descriptionBytes, joined, verified, muted, uint64(requestedToJoinAt.Int64), p.logger)
if err != nil {
return nil, err
}
response = append(response, org)
}
return response, nil
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) AllCommunities(memberIdentity *ecdsa.PublicKey) ([]*Community, error) {
return p.queryCommunities(memberIdentity, communitiesBaseQuery)
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) JoinedCommunities(memberIdentity *ecdsa.PublicKey) ([]*Community, error) {
query := communitiesBaseQuery + ` WHERE c.joined`
return p.queryCommunities(memberIdentity, query)
}
func (p *Persistence) JoinedAndPendingCommunitiesWithRequests(memberIdentity *ecdsa.PublicKey) (comms []*Community, err error) {
query := `SELECT
c.id, c.private_key, c.description, c.joined, c.verified, c.muted,
r.id, r.public_key, r.clock, r.ens_name, r.chat_id, r.community_id, r.state
FROM communities_communities c
LEFT JOIN communities_requests_to_join r ON c.id = r.community_id AND r.public_key = ?
WHERE c.Joined OR r.state = ?`
rows, err := p.db.Query(query, common.PubkeyToHex(memberIdentity), RequestToJoinStatePending)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
// Don't shadow original error
_ = rows.Close()
return
}
err = rows.Close()
}()
for rows.Next() {
var comm *Community
// Community specific fields
var publicKeyBytes, privateKeyBytes, descriptionBytes []byte
var joined, verified, muted bool
// Request to join specific fields
var rtjID, rtjCommunityID []byte
var rtjPublicKey, rtjENSName, rtjChatID sql.NullString
var rtjClock, rtjState sql.NullInt64
err = rows.Scan(
&publicKeyBytes, &privateKeyBytes, &descriptionBytes, &joined, &verified, &muted,
&rtjID, &rtjPublicKey, &rtjClock, &rtjENSName, &rtjChatID, &rtjCommunityID, &rtjState)
if err != nil {
return nil, err
}
comm, err = unmarshalCommunityFromDB(memberIdentity, publicKeyBytes, privateKeyBytes, descriptionBytes, joined, verified, muted, uint64(rtjClock.Int64), p.logger)
if err != nil {
return nil, err
}
rtj := unmarshalRequestToJoinFromDB(rtjID, rtjCommunityID, rtjPublicKey, rtjENSName, rtjChatID, rtjClock, rtjState)
if !rtj.Empty() {
comm.AddRequestToJoin(rtj)
}
comms = append(comms, comm)
}
return comms, nil
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) CreatedCommunities(memberIdentity *ecdsa.PublicKey) ([]*Community, error) {
query := communitiesBaseQuery + ` WHERE c.private_key IS NOT NULL`
return p.queryCommunities(memberIdentity, query)
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) GetByID(memberIdentity *ecdsa.PublicKey, id []byte) (*Community, error) {
var publicKeyBytes, privateKeyBytes, descriptionBytes []byte
var joined bool
var verified bool
var muted bool
2021-01-11 10:32:51 +00:00
var requestedToJoinAt sql.NullInt64
err := p.db.QueryRow(communitiesBaseQuery+` WHERE c.id = ?`, common.PubkeyToHex(memberIdentity), id).Scan(&publicKeyBytes, &privateKeyBytes, &descriptionBytes, &joined, &verified, &muted, &requestedToJoinAt)
if err == sql.ErrNoRows {
return nil, nil
} else if err != nil {
return nil, err
}
return unmarshalCommunityFromDB(memberIdentity, publicKeyBytes, privateKeyBytes, descriptionBytes, joined, verified, muted, uint64(requestedToJoinAt.Int64), p.logger)
}
func unmarshalCommunityFromDB(memberIdentity *ecdsa.PublicKey, publicKeyBytes, privateKeyBytes, descriptionBytes []byte, joined, verified, muted bool, requestedToJoinAt uint64, logger *zap.Logger) (*Community, error) {
var privateKey *ecdsa.PrivateKey
var err error
if privateKeyBytes != nil {
privateKey, err = crypto.ToECDSA(privateKeyBytes)
if err != nil {
return nil, err
}
}
metadata := &protobuf.ApplicationMetadataMessage{}
err = proto.Unmarshal(descriptionBytes, metadata)
if err != nil {
return nil, err
}
description := &protobuf.CommunityDescription{}
err = proto.Unmarshal(metadata.Payload, description)
if err != nil {
return nil, err
}
id, err := crypto.DecompressPubkey(publicKeyBytes)
if err != nil {
return nil, err
}
config := Config{
PrivateKey: privateKey,
CommunityDescription: description,
2021-01-11 10:32:51 +00:00
MemberIdentity: memberIdentity,
MarshaledCommunityDescription: descriptionBytes,
Logger: logger,
ID: id,
Verified: verified,
Muted: muted,
2021-01-11 10:32:51 +00:00
RequestedToJoinAt: requestedToJoinAt,
Joined: joined,
}
return New(config)
}
2021-01-11 10:32:51 +00:00
func unmarshalRequestToJoinFromDB(ID, communityID []byte, publicKey, ensName, chatID sql.NullString, clock, state sql.NullInt64) *RequestToJoin {
return &RequestToJoin{
ID: ID,
PublicKey: publicKey.String,
Clock: uint64(clock.Int64),
ENSName: ensName.String,
ChatID: chatID.String,
CommunityID: communityID,
State: RequestToJoinState(state.Int64),
}
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) SaveRequestToJoin(request *RequestToJoin) (err error) {
tx, err := p.db.BeginTx(context.Background(), &sql.TxOptions{})
if err != nil {
return err
}
defer func() {
if err == nil {
err = tx.Commit()
return
}
// don't shadow original error
_ = tx.Rollback()
}()
var clock uint64
// Fetch any existing request to join
err = tx.QueryRow(`SELECT clock FROM communities_requests_to_join WHERE state = ? AND public_key = ? AND community_id = ?`, RequestToJoinStatePending, request.PublicKey, request.CommunityID).Scan(&clock)
if err != nil && err != sql.ErrNoRows {
return err
}
// This is already processed
if clock >= request.Clock {
return errors.New("old request to join")
}
_, err = tx.Exec(`INSERT INTO communities_requests_to_join(id,public_key,clock,ens_name,chat_id,community_id,state) VALUES (?, ?, ?, ?, ?, ?, ?)`, request.ID, request.PublicKey, request.Clock, request.ENSName, request.ChatID, request.CommunityID, request.State)
return err
}
func (p *Persistence) PendingRequestsToJoinForUser(pk string) ([]*RequestToJoin, error) {
var requests []*RequestToJoin
rows, err := p.db.Query(`SELECT id,public_key,clock,ens_name,chat_id,community_id,state FROM communities_requests_to_join WHERE state = ? AND public_key = ?`, RequestToJoinStatePending, pk)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
request := &RequestToJoin{}
err := rows.Scan(&request.ID, &request.PublicKey, &request.Clock, &request.ENSName, &request.ChatID, &request.CommunityID, &request.State)
if err != nil {
return nil, err
}
requests = append(requests, request)
}
return requests, nil
}
func (p *Persistence) HasPendingRequestsToJoinForUserAndCommunity(userPk string, communityID []byte) (bool, error) {
var count int
err := p.db.QueryRow(`SELECT count(1) FROM communities_requests_to_join WHERE state = ? AND public_key = ? AND community_id = ?`, RequestToJoinStatePending, userPk, communityID).Scan(&count)
if err != nil {
return false, err
}
return count > 0, nil
}
func (p *Persistence) PendingRequestsToJoinForCommunity(id []byte) ([]*RequestToJoin, error) {
var requests []*RequestToJoin
rows, err := p.db.Query(`SELECT id,public_key,clock,ens_name,chat_id,community_id,state FROM communities_requests_to_join WHERE state = ? AND community_id = ?`, RequestToJoinStatePending, id)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
request := &RequestToJoin{}
err := rows.Scan(&request.ID, &request.PublicKey, &request.Clock, &request.ENSName, &request.ChatID, &request.CommunityID, &request.State)
if err != nil {
return nil, err
}
requests = append(requests, request)
}
return requests, nil
}
func (p *Persistence) SetRequestToJoinState(pk string, communityID []byte, state RequestToJoinState) error {
2021-01-11 10:32:51 +00:00
_, err := p.db.Exec(`UPDATE communities_requests_to_join SET state = ? WHERE community_id = ? AND public_key = ?`, state, communityID, pk)
return err
}
func (p *Persistence) SetMuted(communityID []byte, muted bool) error {
_, err := p.db.Exec(`UPDATE communities_communities SET muted = ? WHERE id = ?`, muted, communityID)
return err
}
2021-01-11 10:32:51 +00:00
func (p *Persistence) GetRequestToJoin(id []byte) (*RequestToJoin, error) {
request := &RequestToJoin{}
err := p.db.QueryRow(`SELECT id,public_key,clock,ens_name,chat_id,community_id,state FROM communities_requests_to_join WHERE id = ?`, id).Scan(&request.ID, &request.PublicKey, &request.Clock, &request.ENSName, &request.ChatID, &request.CommunityID, &request.State)
if err != nil {
return nil, err
}
return request, nil
}
func (p *Persistence) SetSyncClock(id []byte, clock uint64) error {
_, err := p.db.Exec(`UPDATE communities_communities SET synced_at = ? WHERE id = ? AND synced_at < ?`, clock, id, clock)
return err
}
func (p *Persistence) SetPrivateKey(id []byte, privKey *ecdsa.PrivateKey) error {
_, err := p.db.Exec(`UPDATE communities_communities SET private_key = ? WHERE id = ?`, crypto.FromECDSA(privKey), id)
return err
}