2021-04-13 18:52:57 +00:00
package persistence
2021-04-12 17:59:41 +00:00
import (
"database/sql"
"log"
2021-04-22 00:09:37 +00:00
"github.com/status-im/go-waku/waku/v2/protocol/pb"
"github.com/status-im/go-waku/waku/v2/protocol/store"
2021-04-12 17:59:41 +00:00
)
2021-04-22 18:49:52 +00:00
// DBStore is a MessageProvider that has a *sql.DB connection
2021-04-12 17:59:41 +00:00
type DBStore struct {
store . MessageProvider
db * sql . DB
}
2021-10-09 18:18:53 +00:00
// DBOption is an optional setting that can be used to configure the DBStore
2021-04-13 18:52:57 +00:00
type DBOption func ( * DBStore ) error
2021-04-22 18:49:52 +00:00
// WithDB is a DBOption that lets you use any custom *sql.DB with a DBStore.
2021-04-13 18:52:57 +00:00
func WithDB ( db * sql . DB ) DBOption {
return func ( d * DBStore ) error {
d . db = db
return nil
}
}
2021-04-22 18:49:52 +00:00
// WithDriver is a DBOption that will open a *sql.DB connection
2021-04-13 18:52:57 +00:00
func WithDriver ( driverName string , datasourceName string ) DBOption {
return func ( d * DBStore ) error {
db , err := sql . Open ( driverName , datasourceName )
if err != nil {
return err
}
d . db = db
return nil
}
}
2021-04-22 18:49:52 +00:00
// Creates a new DB store using the db specified via options.
// It will create a messages table if it does not exist
2021-04-13 18:52:57 +00:00
func NewDBStore ( opt DBOption ) ( * DBStore , error ) {
result := new ( DBStore )
err := opt ( result )
2021-04-12 17:59:41 +00:00
if err != nil {
return nil , err
}
err = result . createTable ( )
if err != nil {
return nil , err
}
return result , nil
}
func ( d * DBStore ) createTable ( ) error {
2021-04-28 15:11:32 +00:00
sqlStmt := ` CREATE TABLE IF NOT EXISTS message (
2021-04-12 17:59:41 +00:00
id BLOB PRIMARY KEY ,
2021-07-29 15:03:30 +00:00
receiverTimestamp REAL NOT NULL ,
senderTimestamp REAL NOT NULL ,
2021-04-12 17:59:41 +00:00
contentTopic BLOB NOT NULL ,
2021-04-28 15:11:32 +00:00
pubsubTopic BLOB NOT NULL ,
2021-04-12 17:59:41 +00:00
payload BLOB ,
version INTEGER NOT NULL DEFAULT 0
) WITHOUT ROWID ; `
_ , err := d . db . Exec ( sqlStmt )
if err != nil {
return err
}
return nil
}
2021-04-22 18:49:52 +00:00
// Closes a DB connection
2021-04-12 17:59:41 +00:00
func ( d * DBStore ) Stop ( ) {
d . db . Close ( )
}
2021-04-22 18:49:52 +00:00
// Inserts a WakuMessage into the DB
2021-04-28 15:11:32 +00:00
func ( d * DBStore ) Put ( cursor * pb . Index , pubsubTopic string , message * pb . WakuMessage ) error {
2021-07-29 15:03:30 +00:00
stmt , err := d . db . Prepare ( "INSERT INTO message (id, receiverTimestamp, senderTimestamp, contentTopic, pubsubTopic, payload, version) VALUES (?, ?, ?, ?, ?, ?, ?)" )
2021-04-12 17:59:41 +00:00
if err != nil {
return err
}
2021-07-29 15:03:30 +00:00
_ , err = stmt . Exec ( cursor . Digest , cursor . ReceiverTime , message . Timestamp , message . ContentTopic , pubsubTopic , message . Payload , message . Version )
2021-04-12 17:59:41 +00:00
if err != nil {
return err
}
return nil
}
2021-04-22 18:49:52 +00:00
// Returns all the stored WakuMessages
2021-07-11 18:11:38 +00:00
func ( d * DBStore ) GetAll ( ) ( [ ] store . StoredMessage , error ) {
2021-07-29 15:03:30 +00:00
rows , err := d . db . Query ( "SELECT id, receiverTimestamp, senderTimestamp, contentTopic, pubsubTopic, payload, version FROM message ORDER BY senderTimestamp ASC" )
2021-04-12 17:59:41 +00:00
if err != nil {
return nil , err
}
2021-07-11 18:11:38 +00:00
var result [ ] store . StoredMessage
2021-04-12 17:59:41 +00:00
defer rows . Close ( )
for rows . Next ( ) {
2021-07-11 18:11:38 +00:00
var id [ ] byte
2021-07-29 15:03:30 +00:00
var receiverTimestamp float64
var senderTimestamp float64
2021-04-12 17:59:41 +00:00
var contentTopic string
var payload [ ] byte
var version uint32
2021-04-28 15:11:32 +00:00
var pubsubTopic string
2021-04-12 17:59:41 +00:00
2021-07-11 18:11:38 +00:00
err = rows . Scan ( & id , & receiverTimestamp , & senderTimestamp , & contentTopic , & pubsubTopic , & payload , & version )
2021-04-12 17:59:41 +00:00
if err != nil {
log . Fatal ( err )
}
2021-04-22 00:09:37 +00:00
msg := new ( pb . WakuMessage )
2021-04-12 17:59:41 +00:00
msg . ContentTopic = contentTopic
msg . Payload = payload
2021-07-29 15:03:30 +00:00
msg . Timestamp = senderTimestamp
2021-04-12 17:59:41 +00:00
msg . Version = version
2021-07-11 18:11:38 +00:00
record := store . StoredMessage {
ID : id ,
PubsubTopic : pubsubTopic ,
ReceiverTime : receiverTimestamp ,
Message : msg ,
}
result = append ( result , record )
2021-04-12 17:59:41 +00:00
}
err = rows . Err ( )
if err != nil {
return nil , err
}
return result , nil
}