diff --git a/indexedDbWorker/channels/channelsIndexedDbWorker.js b/indexedDb/channels/channelsIndexedDbWorker.js
similarity index 93%
rename from indexedDbWorker/channels/channelsIndexedDbWorker.js
rename to indexedDb/channels/channelsIndexedDbWorker.js
index 7c03a409a1b5e62aab21a78dbffa7affa87b53b3..245c80d0368dc04805e9dadfcf95a7700693483b 100644
--- a/indexedDbWorker/channels/channelsIndexedDbWorker.js
+++ b/indexedDb/channels/channelsIndexedDbWorker.js
@@ -8,7 +8,7 @@
importScripts('wasm_exec.js');
const go = new Go();
-const binPath = 'xxdk-indexedDkWorker.wasm'
+const binPath = 'channelsIndexedDkWorker.wasm'
WebAssembly.instantiateStreaming(fetch(binPath), go.importObject).then((result) => {
go.run(result.instance);
}).catch((err) => {
diff --git a/indexedDbWorker/channels/handlers.go b/indexedDb/channels/handlers.go
similarity index 98%
rename from indexedDbWorker/channels/handlers.go
rename to indexedDb/channels/handlers.go
index b4ba0fcbab623fbfab14f3393392150491749729..d9feece8492cc8beaca31c25404a6f8ea10c2c5b 100644
--- a/indexedDbWorker/channels/handlers.go
+++ b/indexedDb/channels/handlers.go
@@ -20,8 +20,8 @@ import (
"gitlab.com/elixxir/crypto/fastRNG"
"gitlab.com/elixxir/crypto/message"
"gitlab.com/elixxir/xxdk-wasm/indexedDb"
- mChannels "gitlab.com/elixxir/xxdk-wasm/indexedDb/channels"
"gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ mChannels "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker/channels"
"gitlab.com/xx_network/crypto/csprng"
"gitlab.com/xx_network/primitives/id"
"time"
@@ -30,7 +30,7 @@ import (
// manager handles the event model and the message handler, which is used to
// send information between the event model and the main thread.
type manager struct {
- mh *indexedDbWorker.MessageHandler
+ mh *indexedDb2.MessageHandler
model channels.EventModel
}
@@ -81,7 +81,7 @@ func (m *manager) newWASMEventModelHandler(data []byte) []byte {
// messageReceivedCallback sends calls to the MessageReceivedCallback in the
// main thread.
//
-// storeEncryptionStatus adhere to the indexedDb.MessageReceivedCallback type.
+// storeEncryptionStatus adhere to the indexedDbWorker.MessageReceivedCallback type.
func (m *manager) messageReceivedCallback(
uuid uint64, channelID *id.ID, update bool) {
// Package parameters for sending
diff --git a/indexedDb/channels/implementation.go b/indexedDb/channels/implementation.go
index 6538e4f76c7d894cce5b8373364b796f78106528..a971ec9e537621371945e29fa87180b7e479b777 100644
--- a/indexedDb/channels/implementation.go
+++ b/indexedDb/channels/implementation.go
@@ -7,22 +7,28 @@
//go:build js && wasm
-package channels
+package main
import (
"crypto/ed25519"
+ "encoding/base64"
"encoding/json"
- "time"
-
"gitlab.com/elixxir/xxdk-wasm/indexedDb"
+ "strings"
+ "sync"
+ "syscall/js"
+ "time"
+ "github.com/hack-pad/go-indexeddb/idb"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/channels"
"gitlab.com/elixxir/client/v4/cmix/rounds"
cryptoBroadcast "gitlab.com/elixxir/crypto/broadcast"
+ cryptoChannel "gitlab.com/elixxir/crypto/channel"
"gitlab.com/elixxir/crypto/message"
+ "gitlab.com/elixxir/xxdk-wasm/utils"
"gitlab.com/xx_network/primitives/id"
)
@@ -30,23 +36,108 @@ import (
// system passed an object that adheres to in order to get events on the
// channel.
type wasmModel struct {
- wh *indexedDb.WorkerHandler
+ db *idb.Database
+ cipher cryptoChannel.Cipher
+ receivedMessageCB MessageReceivedCallback
+ updateMux sync.Mutex
}
// JoinChannel is called whenever a channel is joined locally.
func (w *wasmModel) JoinChannel(channel *cryptoBroadcast.Channel) {
- data, err := json.Marshal(channel)
+ parentErr := errors.New("failed to JoinChannel")
+
+ // Build object
+ newChannel := Channel{
+ ID: channel.ReceptionID.Marshal(),
+ Name: channel.Name,
+ Description: channel.Description,
+ }
+
+ // Convert to jsObject
+ newChannelJson, err := json.Marshal(&newChannel)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to marshal Channel: %+v", err))
+ return
+ }
+ channelObj, err := utils.JsonToJS(newChannelJson)
if err != nil {
- jww.ERROR.Printf("Could not JSON marshal broadcast.Channel: %+v", err)
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to marshal Channel: %+v", err))
return
}
- w.wh.SendMessage(indexedDb.JoinChannelTag, data, nil)
+ _, err = indexedDb2.Put(w.db, channelsStoreName, channelObj)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to put Channel: %+v", err))
+ }
}
// LeaveChannel is called whenever a channel is left locally.
func (w *wasmModel) LeaveChannel(channelID *id.ID) {
- w.wh.SendMessage(indexedDb.LeaveChannelTag, channelID.Marshal(), nil)
+ parentErr := errors.New("failed to LeaveChannel")
+
+ // Delete the channel from storage
+ err := indexedDb2.Delete(w.db, channelsStoreName,
+ js.ValueOf(channelID.String()))
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to delete Channel: %+v", err))
+ return
+ }
+
+ // Clean up lingering data
+ err = w.deleteMsgByChannel(channelID)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Deleting Channel's Message data failed: %+v", err))
+ return
+ }
+ jww.DEBUG.Printf("Successfully deleted channel: %s", channelID)
+}
+
+// deleteMsgByChannel is a private helper that uses messageStoreChannelIndex
+// to delete all Message with the given Channel ID.
+func (w *wasmModel) deleteMsgByChannel(channelID *id.ID) error {
+ parentErr := errors.New("failed to deleteMsgByChannel")
+
+ // Prepare the Transaction
+ txn, err := w.db.Transaction(idb.TransactionReadWrite, messageStoreName)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to create Transaction: %+v", err)
+ }
+ store, err := txn.ObjectStore(messageStoreName)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to get ObjectStore: %+v", err)
+ }
+ index, err := store.Index(messageStoreChannelIndex)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to get Index: %+v", err)
+ }
+
+ // Perform the operation
+ channelIdStr := base64.StdEncoding.EncodeToString(channelID.Marshal())
+ keyRange, err := idb.NewKeyRangeOnly(js.ValueOf(channelIdStr))
+ cursorRequest, err := index.OpenCursorRange(keyRange, idb.CursorNext)
+ if err != nil {
+ return errors.WithMessagef(parentErr, "Unable to open Cursor: %+v", err)
+ }
+ ctx, cancel := indexedDb2.NewContext()
+ err = cursorRequest.Iter(ctx,
+ func(cursor *idb.CursorWithValue) error {
+ _, err := cursor.Delete()
+ return err
+ })
+ cancel()
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to delete Message data: %+v", err)
+ }
+ return nil
}
// ReceiveMessage is called whenever a message is received on a given channel.
@@ -57,58 +148,30 @@ func (w *wasmModel) ReceiveMessage(channelID *id.ID, messageID message.ID,
nickname, text string, pubKey ed25519.PublicKey, dmToken uint32,
codeset uint8, timestamp time.Time, lease time.Duration, round rounds.Round,
mType channels.MessageType, status channels.SentStatus, hidden bool) uint64 {
- msg := channels.ModelMessage{
- Nickname: nickname,
- MessageID: messageID,
- ChannelID: channelID,
- Timestamp: timestamp,
- Lease: lease,
- Status: status,
- Hidden: hidden,
- Pinned: false,
- Content: []byte(text),
- Type: mType,
- Round: round.ID,
- PubKey: pubKey,
- CodesetVersion: codeset,
- DmToken: dmToken,
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for ReceiveMessage: %+v", err)
- return 0
- }
+ textBytes := []byte(text)
+ var err error
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveMessageTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt([]byte(text))
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveMessage: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
-
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveMessage", indexedDb.ResponseTimeout)
}
- return 0
-}
+ msgToInsert := buildMessage(
+ channelID.Marshal(), messageID.Bytes(), nil, nickname,
+ textBytes, pubKey, dmToken, codeset, timestamp, lease, round.ID, mType,
+ false, hidden, status)
+
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ }
-// ReceiveReplyMessage is JSON marshalled and sent to the worker for
-// [wasmModel.ReceiveReply].
-type ReceiveReplyMessage struct {
- ReactionTo message.ID `json:"replyTo"`
- channels.ModelMessage `json:"message"`
+ go w.receivedMessageCB(uuid, channelID, false)
+ return uuid
}
// ReceiveReply is called whenever a message is received that is a reply on a
@@ -122,54 +185,29 @@ func (w *wasmModel) ReceiveReply(channelID *id.ID, messageID,
dmToken uint32, codeset uint8, timestamp time.Time, lease time.Duration,
round rounds.Round, mType channels.MessageType, status channels.SentStatus,
hidden bool) uint64 {
- msg := ReceiveReplyMessage{
- ReactionTo: replyTo,
- ModelMessage: channels.ModelMessage{
- Nickname: nickname,
- MessageID: messageID,
- ChannelID: channelID,
- Timestamp: timestamp,
- Lease: lease,
- Status: status,
- Hidden: hidden,
- Pinned: false,
- Content: []byte(text),
- Type: mType,
- Round: round.ID,
- PubKey: pubKey,
- CodesetVersion: codeset,
- DmToken: dmToken,
- },
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for ReceiveReply: %+v", err)
- return 0
- }
+ textBytes := []byte(text)
+ var err error
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveReplyTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt([]byte(text))
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveReply: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
-
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- return 0
+ msgToInsert := buildMessage(channelID.Marshal(), messageID.Bytes(),
+ replyTo.Bytes(), nickname, textBytes, pubKey, dmToken, codeset,
+ timestamp, lease, round.ID, mType, hidden, false, status)
+
+ uuid, err := w.receiveHelper(msgToInsert, false)
+
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive reply: %+v", err)
+ }
+ go w.receivedMessageCB(uuid, channelID, false)
+ return uuid
}
// ReceiveReaction is called whenever a reaction to a message is received on a
@@ -183,79 +221,29 @@ func (w *wasmModel) ReceiveReaction(channelID *id.ID, messageID,
dmToken uint32, codeset uint8, timestamp time.Time, lease time.Duration,
round rounds.Round, mType channels.MessageType, status channels.SentStatus,
hidden bool) uint64 {
+ textBytes := []byte(reaction)
+ var err error
- msg := ReceiveReplyMessage{
- ReactionTo: reactionTo,
- ModelMessage: channels.ModelMessage{
- Nickname: nickname,
- MessageID: messageID,
- ChannelID: channelID,
- Timestamp: timestamp,
- Lease: lease,
- Status: status,
- Hidden: hidden,
- Pinned: false,
- Content: []byte(reaction),
- Type: mType,
- Round: round.ID,
- PubKey: pubKey,
- CodesetVersion: codeset,
- DmToken: dmToken,
- },
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for ReceiveReaction: %+v", err)
- return 0
- }
-
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveReactionTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt([]byte(reaction))
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveReaction: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
-
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- return 0
-}
-
-// MessageUpdateInfo is JSON marshalled and sent to the worker for
-// [wasmModel.UpdateFromMessageID] and [wasmModel.UpdateFromUUID].
-type MessageUpdateInfo struct {
- UUID uint64 `json:"uuid"`
-
- MessageID message.ID `json:"messageID"`
- MessageIDSet bool `json:"messageIDSet"`
-
- Timestamp time.Time `json:"timestamp"`
- TimestampSet bool `json:"timestampSet"`
+ msgToInsert := buildMessage(
+ channelID.Marshal(), messageID.Bytes(), reactionTo.Bytes(), nickname,
+ textBytes, pubKey, dmToken, codeset, timestamp, lease, round.ID, mType,
+ false, hidden, status)
- RoundID id.Round `json:"round"`
- RoundIDSet bool `json:"roundIDSet"`
-
- Pinned bool `json:"pinned"`
- PinnedSet bool `json:"pinnedSet"`
-
- Hidden bool `json:"hidden"`
- HiddenSet bool `json:"hiddenSet"`
-
- Status channels.SentStatus `json:"status"`
- StatusSet bool `json:"statusSet"`
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive reaction: %+v", err)
+ }
+ go w.receivedMessageCB(uuid, channelID, false)
+ return uuid
}
// UpdateFromUUID is called whenever a message at the UUID is modified.
@@ -266,40 +254,31 @@ type MessageUpdateInfo struct {
func (w *wasmModel) UpdateFromUUID(uuid uint64, messageID *message.ID,
timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
status *channels.SentStatus) {
- msg := MessageUpdateInfo{UUID: uuid}
- if messageID != nil {
- msg.MessageID = *messageID
- msg.MessageIDSet = true
- }
- if timestamp != nil {
- msg.Timestamp = *timestamp
- msg.TimestampSet = true
- }
- if round != nil {
- msg.RoundID = round.ID
- msg.RoundIDSet = true
- }
- if pinned != nil {
- msg.Pinned = *pinned
- msg.PinnedSet = true
- }
- if hidden != nil {
- msg.Hidden = *hidden
- msg.HiddenSet = true
- }
- if status != nil {
- msg.Status = *status
- msg.StatusSet = true
- }
+ parentErr := errors.New("failed to UpdateFromUUID")
- data, err := json.Marshal(msg)
+ // FIXME: this is a bit of race condition without the mux.
+ // This should be done via the transactions (i.e., make a
+ // special version of receiveHelper)
+ w.updateMux.Lock()
+ defer w.updateMux.Unlock()
+
+ // Convert messageID to the key generated by json.Marshal
+ key := js.ValueOf(uuid)
+
+ // Use the key to get the existing Message
+ currentMsg, err := indexedDb2.Get(w.db, messageStoreName, key)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for UpdateFromUUID: %+v", err)
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get message: %+v", err))
return
}
- w.wh.SendMessage(indexedDb.UpdateFromUUIDTag, data, nil)
+ _, err = w.updateMessage(utils.JsToJson(currentMsg), messageID, timestamp,
+ round, pinned, hidden, status)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to updateMessage: %+v", err))
+ }
}
// UpdateFromMessageID is called whenever a message with the message ID is
@@ -314,109 +293,222 @@ func (w *wasmModel) UpdateFromUUID(uuid uint64, messageID *message.ID,
func (w *wasmModel) UpdateFromMessageID(messageID message.ID,
timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
status *channels.SentStatus) uint64 {
+ parentErr := errors.New("failed to UpdateFromMessageID")
- msg := MessageUpdateInfo{MessageID: messageID, MessageIDSet: true}
- if timestamp != nil {
- msg.Timestamp = *timestamp
- msg.TimestampSet = true
+ // FIXME: this is a bit of race condition without the mux.
+ // This should be done via the transactions (i.e., make a
+ // special version of receiveHelper)
+ w.updateMux.Lock()
+ defer w.updateMux.Unlock()
+
+ msgIDStr := base64.StdEncoding.EncodeToString(messageID.Marshal())
+ currentMsgObj, err := indexedDb2.GetIndex(w.db, messageStoreName,
+ messageStoreMessageIndex, js.ValueOf(msgIDStr))
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Failed to get message by index: %+v", err))
+ return 0
}
+
+ currentMsg := utils.JsToJson(currentMsgObj)
+ uuid, err := w.updateMessage(currentMsg, &messageID, timestamp,
+ round, pinned, hidden, status)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to updateMessage: %+v", err))
+ }
+ return uuid
+}
+
+// updateMessage is a helper for updating a stored message.
+func (w *wasmModel) updateMessage(currentMsgJson string, messageID *message.ID,
+ timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
+ status *channels.SentStatus) (uint64, error) {
+
+ newMessage := &Message{}
+ err := json.Unmarshal([]byte(currentMsgJson), newMessage)
+ if err != nil {
+ return 0, err
+ }
+
+ if status != nil {
+ newMessage.Status = uint8(*status)
+ }
+ if messageID != nil {
+ newMessage.MessageID = messageID.Bytes()
+ }
+
if round != nil {
- msg.RoundID = round.ID
- msg.RoundIDSet = true
+ newMessage.Round = uint64(round.ID)
+ }
+
+ if timestamp != nil {
+ newMessage.Timestamp = *timestamp
}
+
if pinned != nil {
- msg.Pinned = *pinned
- msg.PinnedSet = true
+ newMessage.Pinned = *pinned
}
+
if hidden != nil {
- msg.Hidden = *hidden
- msg.HiddenSet = true
+ newMessage.Hidden = *hidden
}
- if status != nil {
- msg.Status = *status
- msg.StatusSet = true
+
+ // Store the updated Message
+ uuid, err := w.receiveHelper(newMessage, true)
+ if err != nil {
+ return 0, err
+ }
+ channelID := &id.ID{}
+ copy(channelID[:], newMessage.ChannelID)
+ go w.receivedMessageCB(uuid, channelID, true)
+
+ return uuid, nil
+}
+
+// buildMessage is a private helper that converts typical [channels.EventModel]
+// inputs into a basic Message structure for insertion into storage.
+//
+// NOTE: ID is not set inside this function because we want to use the
+// autoincrement key by default. If you are trying to overwrite an existing
+// message, then you need to set it manually yourself.
+func buildMessage(channelID, messageID, parentID []byte, nickname string,
+ text []byte, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
+ timestamp time.Time, lease time.Duration, round id.Round,
+ mType channels.MessageType, pinned, hidden bool,
+ status channels.SentStatus) *Message {
+ return &Message{
+ MessageID: messageID,
+ Nickname: nickname,
+ ChannelID: channelID,
+ ParentMessageID: parentID,
+ Timestamp: timestamp,
+ Lease: lease,
+ Status: uint8(status),
+ Hidden: hidden,
+ Pinned: pinned,
+ Text: text,
+ Type: uint16(mType),
+ Round: uint64(round),
+ // User Identity Info
+ Pubkey: pubKey,
+ DmToken: dmToken,
+ CodesetVersion: codeset,
}
+}
- data, err := json.Marshal(msg)
+// receiveHelper is a private helper for receiving any sort of message.
+func (w *wasmModel) receiveHelper(
+ newMessage *Message, isUpdate bool) (uint64, error) {
+ // Convert to jsObject
+ newMessageJson, err := json.Marshal(newMessage)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for UpdateFromMessageID: %+v", err)
- return 0
+ return 0, errors.Errorf("Unable to marshal Message: %+v", err)
+ }
+ messageObj, err := utils.JsonToJS(newMessageJson)
+ if err != nil {
+ return 0, errors.Errorf("Unable to marshal Message: %+v", err)
}
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.UpdateFromMessageIDTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to UpdateFromMessageID: %+v", err)
- uuidChan <- 0
- }
- uuidChan <- uuid
- })
+ // Unset the primaryKey for inserts so that it can be auto-populated and
+ // incremented
+ if !isUpdate {
+ messageObj.Delete("id")
+ }
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveReply", indexedDb.ResponseTimeout)
+ // Store message to database
+ result, err := indexedDb2.Put(w.db, messageStoreName, messageObj)
+ if err != nil && !strings.Contains(err.Error(),
+ "at least one key does not satisfy the uniqueness requirements") {
+ // Only return non-unique constraint errors so that the case
+ // below this one can be hit and handle duplicate entries properly.
+ return 0, errors.Errorf("Unable to put Message: %+v", err)
}
- return 0
-}
+ // NOTE: Sometimes the insert fails to return an error but hits a duplicate
+ // insert, so this fallthrough returns the UUID entry in that case.
+ if result.IsUndefined() {
+ msgID := message.ID{}
+ copy(msgID[:], newMessage.MessageID)
+ msg, errLookup := w.msgIDLookup(msgID)
+ if errLookup == nil && msg.ID != 0 {
+ return msg.ID, nil
+ }
+ return 0, errors.Errorf("uuid lookup failure: %+v", err)
+ }
+ uuid := uint64(result.Int())
+ jww.DEBUG.Printf("Successfully stored message %d", uuid)
-// GetMessageMessage is JSON marshalled and sent to the worker for
-// [wasmModel.GetMessage].
-type GetMessageMessage struct {
- Message channels.ModelMessage `json:"message"`
- Error string `json:"error"`
+ return uuid, nil
}
// GetMessage returns the message with the given [channel.MessageID].
func (w *wasmModel) GetMessage(
messageID message.ID) (channels.ModelMessage, error) {
- msgChan := make(chan GetMessageMessage)
- w.wh.SendMessage(indexedDb.GetMessageTag, messageID.Marshal(), func(data []byte) {
- var msg GetMessageMessage
- err := json.Unmarshal(data, &msg)
+ lookupResult, err := w.msgIDLookup(messageID)
+ if err != nil {
+ return channels.ModelMessage{}, err
+ }
+
+ var channelId *id.ID
+ if lookupResult.ChannelID != nil {
+ channelId, err = id.Unmarshal(lookupResult.ChannelID)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to GetMessage: %+v", err)
+ return channels.ModelMessage{}, err
}
- msgChan <- msg
- })
+ }
- select {
- case msg := <-msgChan:
- if msg.Error != "" {
- return channels.ModelMessage{}, errors.New(msg.Error)
+ var parentMsgId message.ID
+ if lookupResult.ParentMessageID != nil {
+ parentMsgId, err = message.UnmarshalID(lookupResult.ParentMessageID)
+ if err != nil {
+ return channels.ModelMessage{}, err
}
- return msg.Message, nil
- case <-time.After(indexedDb.ResponseTimeout):
- return channels.ModelMessage{}, errors.Errorf("timed out after %s "+
- "waiting for response from the worker about GetMessage",
- indexedDb.ResponseTimeout)
}
+
+ return channels.ModelMessage{
+ UUID: lookupResult.ID,
+ Nickname: lookupResult.Nickname,
+ MessageID: messageID,
+ ChannelID: channelId,
+ ParentMessageID: parentMsgId,
+ Timestamp: lookupResult.Timestamp,
+ Lease: lookupResult.Lease,
+ Status: channels.SentStatus(lookupResult.Status),
+ Hidden: lookupResult.Hidden,
+ Pinned: lookupResult.Pinned,
+ Content: lookupResult.Text,
+ Type: channels.MessageType(lookupResult.Type),
+ Round: id.Round(lookupResult.Round),
+ PubKey: lookupResult.Pubkey,
+ CodesetVersion: lookupResult.CodesetVersion,
+ }, nil
}
// DeleteMessage removes a message with the given messageID from storage.
func (w *wasmModel) DeleteMessage(messageID message.ID) error {
- errChan := make(chan error)
- w.wh.SendMessage(indexedDb.DeleteMessageTag, messageID.Marshal(), func(data []byte) {
- if data != nil {
- errChan <- errors.New(string(data))
- } else {
- errChan <- nil
- }
- })
+ msgId := js.ValueOf(base64.StdEncoding.EncodeToString(messageID.Bytes()))
+ return indexedDb2.DeleteIndex(w.db, messageStoreName,
+ messageStoreMessageIndex, pkeyName, msgId)
+}
+
+// msgIDLookup gets the UUID of the Message with the given messageID.
+func (w *wasmModel) msgIDLookup(messageID message.ID) (*Message, error) {
+ msgIDStr := js.ValueOf(base64.StdEncoding.EncodeToString(messageID.Bytes()))
+ resultObj, err := indexedDb2.GetIndex(w.db, messageStoreName,
+ messageStoreMessageIndex, msgIDStr)
+ if err != nil {
+ return nil, err
+ } else if resultObj.IsUndefined() {
+ return nil, errors.Errorf("no message for %s found", msgIDStr)
+ }
- select {
- case err := <-errChan:
- return err
- case <-time.After(indexedDb.ResponseTimeout):
- return errors.Errorf("timed out after %s waiting for response from "+
- "the worker about DeleteMessage", indexedDb.ResponseTimeout)
+ // Process result into string
+ resultMsg := &Message{}
+ err = json.Unmarshal([]byte(utils.JsToJson(resultObj)), resultMsg)
+ if err != nil {
+ return nil, err
}
+ return resultMsg, nil
+
}
diff --git a/indexedDbWorker/channels/implementation_test.go b/indexedDb/channels/implementation_test.go
similarity index 95%
rename from indexedDbWorker/channels/implementation_test.go
rename to indexedDb/channels/implementation_test.go
index 1a78e4702848e71d5b50b132dfacf018836360f9..22e59146634d68fadf9c6b808bb0cd406eb38caa 100644
--- a/indexedDbWorker/channels/implementation_test.go
+++ b/indexedDb/channels/implementation_test.go
@@ -14,7 +14,7 @@ import (
"fmt"
"github.com/hack-pad/go-indexeddb/idb"
"gitlab.com/elixxir/crypto/message"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
"gitlab.com/elixxir/xxdk-wasm/storage"
"gitlab.com/xx_network/crypto/csprng"
"gitlab.com/xx_network/primitives/netTime"
@@ -107,7 +107,7 @@ func TestWasmModel_DeleteMessage(t *testing.T) {
}
// Check the resulting status
- results, err := indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ results, err := indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -122,7 +122,7 @@ func TestWasmModel_DeleteMessage(t *testing.T) {
}
// Check the resulting status
- results, err = indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ results, err = indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -164,7 +164,7 @@ func Test_wasmModel_UpdateSentStatus(t *testing.T) {
}
// Ensure one message is stored
- results, err := indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ results, err := indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -178,7 +178,7 @@ func Test_wasmModel_UpdateSentStatus(t *testing.T) {
uuid, nil, nil, nil, nil, nil, &expectedStatus)
// Check the resulting status
- results, err = indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ results, err = indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -236,7 +236,7 @@ func Test_wasmModel_JoinChannel_LeaveChannel(t *testing.T) {
}
eventModel.JoinChannel(testChannel)
eventModel.JoinChannel(testChannel2)
- results, err := indexedDbWorker.Dump(eventModel.db, channelsStoreName)
+ results, err := indexedDb2.Dump(eventModel.db, channelsStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -244,7 +244,7 @@ func Test_wasmModel_JoinChannel_LeaveChannel(t *testing.T) {
t.Fatalf("Expected 2 channels to exist")
}
eventModel.LeaveChannel(testChannel.ReceptionID)
- results, err = indexedDbWorker.Dump(eventModel.db, channelsStoreName)
+ results, err = indexedDb2.Dump(eventModel.db, channelsStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -396,7 +396,7 @@ func Test_wasmModel_deleteMsgByChannel(t *testing.T) {
}
// Check pre-results
- result, err := indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ result, err := indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
@@ -411,7 +411,7 @@ func Test_wasmModel_deleteMsgByChannel(t *testing.T) {
}
// Check final results
- result, err = indexedDbWorker.Dump(eventModel.db, messageStoreName)
+ result, err = indexedDb2.Dump(eventModel.db, messageStoreName)
if err != nil {
t.Fatalf("%+v", err)
}
diff --git a/indexedDb/channels/init.go b/indexedDb/channels/init.go
index 20f5108def2a3f88e200a1841036836d19d81e67..b9f63e391bbe9409312922ed9a8a621e7d0577db 100644
--- a/indexedDb/channels/init.go
+++ b/indexedDb/channels/init.go
@@ -7,168 +7,248 @@
//go:build js && wasm
-package channels
+package main
import (
- "encoding/json"
+ "github.com/hack-pad/go-indexeddb/idb"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/channels"
cryptoChannel "gitlab.com/elixxir/crypto/channel"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
"gitlab.com/elixxir/xxdk-wasm/indexedDb"
- "gitlab.com/elixxir/xxdk-wasm/storage"
"gitlab.com/xx_network/primitives/id"
+ "syscall/js"
"time"
)
-// WorkerJavascriptFileURL is the URL of the script the worker will execute to
-// launch the worker WASM binary. It must obey the same-origin policy.
-const WorkerJavascriptFileURL = "/integrations/assets/dmIndexedDbWorker.js"
+const (
+ // databaseSuffix is the suffix to be appended to the name of
+ // the database.
+ databaseSuffix = "_speakeasy"
+
+ // currentVersion is the current version of the IndexDb
+ // runtime. Used for migration purposes.
+ currentVersion uint = 1
+)
// MessageReceivedCallback is called any time a message is received or updated.
//
// update is true if the row is old and was edited.
type MessageReceivedCallback func(uuid uint64, channelID *id.ID, update bool)
-// NewWASMEventModelBuilder returns an EventModelBuilder which allows
-// the channel manager to define the path but the callback is the same
-// across the board.
-func NewWASMEventModelBuilder(encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback) channels.EventModelBuilder {
- fn := func(path string) (channels.EventModel, error) {
- return NewWASMEventModel(path, encryption, cb)
- }
- return fn
-}
-
-// NewWASMEventModelMessage is JSON marshalled and sent to the worker for
-// [NewWASMEventModel].
-type NewWASMEventModelMessage struct {
- Path string `json:"path"`
- EncryptionJSON string `json:"encryptionJSON"`
-}
-
// NewWASMEventModel returns a [channels.EventModel] backed by a wasmModel.
// The name should be a base64 encoding of the users public key.
func NewWASMEventModel(path string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback) (channels.EventModel, error) {
+ cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
+ channels.EventModel, error) {
+ databaseName := path + databaseSuffix
+ return newWASMModel(databaseName, encryption, cb, storeEncryptionStatus)
+}
- // TODO: bring in URL and name from caller
- wh, err := indexedDb.NewWorkerHandler(
- WorkerJavascriptFileURL, "channelsIndexedDb")
+// storeEncryptionStatusFn matches storage.StoreIndexedDbEncryptionStatus so
+// that the data can be sent between the worker and main thread.
+type storeEncryptionStatusFn func(
+ databaseName string, encryptionStatus bool) (bool, error)
+
+// newWASMModel creates the given [idb.Database] and returns a wasmModel.
+func newWASMModel(databaseName string, encryption cryptoChannel.Cipher,
+ cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
+ *wasmModel, error) {
+ // Attempt to open database object
+ ctx, cancel := indexedDb2.NewContext()
+ defer cancel()
+ openRequest, err := idb.Global().Open(ctx, databaseName, currentVersion,
+ func(db *idb.Database, oldVersion, newVersion uint) error {
+ if oldVersion == newVersion {
+ jww.INFO.Printf("IndexDb version is current: v%d",
+ newVersion)
+ return nil
+ }
+
+ jww.INFO.Printf("IndexDb upgrade required: v%d -> v%d",
+ oldVersion, newVersion)
+
+ if oldVersion == 0 && newVersion >= 1 {
+ err := v1Upgrade(db)
+ if err != nil {
+ return err
+ }
+ oldVersion = 1
+ }
+
+ // if oldVersion == 1 && newVersion >= 2 { v2Upgrade(), oldVersion = 2 }
+ return nil
+ })
if err != nil {
return nil, err
}
- // Register handler to manage messages for the MessageReceivedCallback
- wh.RegisterHandler(indexedDb.GetMessageTag, indexedDb.InitID, false,
- messageReceivedCallbackHandler(cb))
-
- // Register handler to manage checking encryption status from local storage
- wh.RegisterHandler(indexedDb.EncryptionStatusTag, indexedDb.InitID, false,
- checkDbEncryptionStatusHandler(wh))
+ // Wait for database open to finish
+ db, err := openRequest.Await(ctx)
+ if err != nil {
+ return nil, err
+ }
- encryptionJSON, err := json.Marshal(encryption)
+ // Save the encryption status to storage
+ encryptionStatus := encryption != nil
+ loadedEncryptionStatus, err :=
+ storeEncryptionStatus(databaseName, encryptionStatus)
if err != nil {
return nil, err
}
- message := NewWASMEventModelMessage{
- Path: path,
- EncryptionJSON: string(encryptionJSON),
+ // Verify encryption status does not change
+ if encryptionStatus != loadedEncryptionStatus {
+ return nil, errors.New(
+ "Cannot load database with different encryption status.")
+ } else if !encryptionStatus {
+ jww.WARN.Printf("IndexedDb encryption disabled!")
}
- payload, err := json.Marshal(message)
+ // Attempt to ensure the database has been properly initialized
+ openRequest, err = idb.Global().Open(ctx, databaseName, currentVersion,
+ func(db *idb.Database, oldVersion, newVersion uint) error {
+ return nil
+ })
if err != nil {
return nil, err
}
+ // Wait for database open to finish
+ db, err = openRequest.Await(ctx)
+ if err != nil {
+ return nil, err
+ }
+ wrapper := &wasmModel{db: db, receivedMessageCB: cb, cipher: encryption}
- errChan := make(chan string)
- wh.SendMessage(indexedDb.NewWASMEventModelTag, payload, func(data []byte) {
- errChan <- string(data)
- })
+ return wrapper, nil
+}
- select {
- case workerErr := <-errChan:
- if workerErr != "" {
- return nil, errors.New(workerErr)
- }
- case <-time.After(indexedDb.ResponseTimeout):
- return nil, errors.Errorf("timed out after %s waiting for indexedDB "+
- "database in worker to intialize", indexedDb.ResponseTimeout)
+// v1Upgrade performs the v0 -> v1 database upgrade.
+//
+// This can never be changed without permanently breaking backwards
+// compatibility.
+func v1Upgrade(db *idb.Database) error {
+ storeOpts := idb.ObjectStoreOptions{
+ KeyPath: js.ValueOf(pkeyName),
+ AutoIncrement: true,
+ }
+ indexOpts := idb.IndexOptions{
+ Unique: false,
+ MultiEntry: false,
}
- return &wasmModel{wh}, nil
-}
+ // Build Message ObjectStore and Indexes
+ messageStore, err := db.CreateObjectStore(messageStoreName, storeOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreMessageIndex,
+ js.ValueOf(messageStoreMessage),
+ idb.IndexOptions{
+ Unique: true,
+ MultiEntry: false,
+ })
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreChannelIndex,
+ js.ValueOf(messageStoreChannel), indexOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreParentIndex,
+ js.ValueOf(messageStoreParent), indexOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreTimestampIndex,
+ js.ValueOf(messageStoreTimestamp), indexOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStorePinnedIndex,
+ js.ValueOf(messageStorePinned), indexOpts)
+ if err != nil {
+ return err
+ }
-// MessageReceivedCallbackMessage is JSON marshalled and received from the
-// worker for the [MessageReceivedCallback] callback.
-type MessageReceivedCallbackMessage struct {
- UUID uint64 `json:"uuid"`
- ChannelID *id.ID `json:"channelID"`
- Update bool `json:"update"`
-}
+ // Build Channel ObjectStore
+ _, err = db.CreateObjectStore(channelsStoreName, storeOpts)
+ if err != nil {
+ return err
+ }
-// messageReceivedCallbackHandler returns a handler to manage messages for the
-// MessageReceivedCallback.
-func messageReceivedCallbackHandler(cb MessageReceivedCallback) func(data []byte) {
- return func(data []byte) {
- var msg MessageReceivedCallbackMessage
- err := json.Unmarshal(data, &msg)
- if err != nil {
- jww.ERROR.Printf("Failed to JSON unmarshal "+
- "MessageReceivedCallback message from worker: %+v", err)
- return
- }
- cb(msg.UUID, msg.ChannelID, msg.Update)
+ // Get the database name and save it to storage
+ if databaseName, err2 := db.Name(); err2 != nil {
+ return err2
+ } else if err = storeDatabaseName(databaseName); err != nil {
+ return err
}
-}
-// EncryptionStatusMessage is JSON marshalled and received from the worker when
-// the database checks if it is encrypted.
-type EncryptionStatusMessage struct {
- DatabaseName string `json:"databaseName"`
- EncryptionStatus bool `json:"encryptionStatus"`
+ return nil
}
-// EncryptionStatusReply is JSON marshalled and sent to the worker is response
-// to the [EncryptionStatusMessage].
-type EncryptionStatusReply struct {
- EncryptionStatus bool `json:"encryptionStatus"`
- Error string `json:"error"`
+// hackTestDb is a horrible function that exists as the result of an extremely
+// long discussion about why initializing the IndexedDb sometimes silently
+// fails. It ultimately tries to prevent an unrecoverable situation by actually
+// inserting some nonsense data and then checking to see if it persists.
+// If this function still exists in 2023, god help us all. Amen.
+func (w *wasmModel) hackTestDb() error {
+ testMessage := &Message{
+ ID: 0,
+ Nickname: "test",
+ MessageID: id.DummyUser.Marshal(),
+ }
+ msgId, helper := w.receiveHelper(testMessage, false)
+ if helper != nil {
+ return helper
+ }
+ result, err := indexedDb2.Get(w.db, messageStoreName, js.ValueOf(msgId))
+ if err != nil {
+ return err
+ }
+ if result.IsUndefined() {
+ return errors.Errorf("Failed to test db, record not present")
+ }
+ return nil
}
-// checkDbEncryptionStatusHandler returns a handler to manage checking
-// encryption status from local storage.
-func checkDbEncryptionStatusHandler(wh *indexedDb.WorkerHandler) func(data []byte) {
- return func(data []byte) {
- // Unmarshal received message
- var msg EncryptionStatusMessage
- err := json.Unmarshal(data, &msg)
- if err != nil {
- jww.ERROR.Printf("Failed to JSON unmarshal "+
- "EncryptionStatusMessage message from worker: %+v", err)
- return
- }
-
- // Pass message values to storage
- loadedEncryptionStatus, err := storage.StoreIndexedDbEncryptionStatus(
- msg.DatabaseName, msg.EncryptionStatus)
- var reply EncryptionStatusReply
- if err != nil {
- reply.Error = err.Error()
- } else {
- reply.EncryptionStatus = loadedEncryptionStatus
- }
+// storeDatabaseName sends the database name to storage.StoreIndexedDb in the
+// main thread to be stored in localstorage and waits for the error to be
+// returned.
+//
+// The function specified below is a placeholder until set by
+// registerDatabaseNameStore. registerDatabaseNameStore must be called before
+// storeDatabaseName.
+var storeDatabaseName = func(databaseName string) error { return nil }
+
+// RegisterDatabaseNameStore sets storeDatabaseName to send the database to
+// storage.StoreIndexedDb in the main thread when called and registers a handler
+// to listen for the response.
+func RegisterDatabaseNameStore(m *manager) {
+ storeDatabaseNameResponseChan := make(chan []byte)
+ // Register handler
+ m.mh.RegisterHandler(indexedDb.StoreDatabaseNameTag, func(data []byte) []byte {
+ storeDatabaseNameResponseChan <- data
+ return nil
+ })
- // Return response
- statusData, err := json.Marshal(reply)
- if err != nil {
- jww.ERROR.Printf(
- "Failed to JSON marshal EncryptionStatusReply: %+v", err)
- return
+ storeDatabaseName = func(databaseName string) error {
+ m.mh.SendResponse(indexedDb.StoreDatabaseNameTag, indexedDb.InitID,
+ []byte(databaseName))
+
+ // Wait for response
+ select {
+ case response := <-storeDatabaseNameResponseChan:
+ if len(response) > 0 {
+ return errors.New(string(response))
+ }
+ case <-time.After(indexedDb.ResponseTimeout):
+ return errors.Errorf("timed out after %s waiting for "+
+ "response about storing the database name in local "+
+ "storage in the main thread", indexedDb.ResponseTimeout)
}
-
- wh.SendMessage(indexedDb.EncryptionStatusTag, statusData, nil)
+ return nil
}
}
diff --git a/indexedDbWorker/channels/main.go b/indexedDb/channels/main.go
similarity index 87%
rename from indexedDbWorker/channels/main.go
rename to indexedDb/channels/main.go
index d720ddee52b525d98d1f6027927dd8b750182054..ebe19df08514fc82ebb0543253b51e78435887ab 100644
--- a/indexedDbWorker/channels/main.go
+++ b/indexedDb/channels/main.go
@@ -11,15 +11,15 @@ package main
import (
"fmt"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
)
func main() {
fmt.Println("Starting xxDK WebAssembly Database Worker.")
- m := &manager{mh: indexedDbWorker.NewMessageHandler()}
- m.RegisterHandlers()
+ m := &manager{mh: indexedDb2.NewMessageHandler()}
RegisterDatabaseNameStore(m)
+ m.RegisterHandlers()
m.mh.SignalReady()
<-make(chan bool)
}
diff --git a/indexedDbWorker/channels/model.go b/indexedDb/channels/model.go
similarity index 100%
rename from indexedDbWorker/channels/model.go
rename to indexedDb/channels/model.go
diff --git a/indexedDbWorker/dm/dmIndexedDbWorker.js b/indexedDb/dm/dmIndexedDbWorker.js
similarity index 94%
rename from indexedDbWorker/dm/dmIndexedDbWorker.js
rename to indexedDb/dm/dmIndexedDbWorker.js
index 7c03a409a1b5e62aab21a78dbffa7affa87b53b3..38cd0b5970765048a643b683c6a83d3c670f0888 100644
--- a/indexedDbWorker/dm/dmIndexedDbWorker.js
+++ b/indexedDb/dm/dmIndexedDbWorker.js
@@ -8,7 +8,7 @@
importScripts('wasm_exec.js');
const go = new Go();
-const binPath = 'xxdk-indexedDkWorker.wasm'
+const binPath = 'dmIndexedDbWorker.wasm'
WebAssembly.instantiateStreaming(fetch(binPath), go.importObject).then((result) => {
go.run(result.instance);
}).catch((err) => {
diff --git a/indexedDbWorker/dm/handlers.go b/indexedDb/dm/handlers.go
similarity index 98%
rename from indexedDbWorker/dm/handlers.go
rename to indexedDb/dm/handlers.go
index 52bf98e549b7d3fe0233f5096c27115884566d13..9dbec0990fd1b59bfe0dc0405a323992495c6c0f 100644
--- a/indexedDbWorker/dm/handlers.go
+++ b/indexedDb/dm/handlers.go
@@ -17,10 +17,10 @@ import (
"gitlab.com/elixxir/client/v4/dm"
cryptoChannel "gitlab.com/elixxir/crypto/channel"
"gitlab.com/elixxir/crypto/fastRNG"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
- mChannels "gitlab.com/elixxir/xxdk-wasm/indexedDb/channels"
- mDm "gitlab.com/elixxir/xxdk-wasm/indexedDb/dm"
"gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ mChannels "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker/channels"
+ mDm "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker/dm"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
"gitlab.com/xx_network/crypto/csprng"
"time"
)
@@ -28,7 +28,7 @@ import (
// manager handles the event model and the message handler, which is used to
// send information between the event model and the main thread.
type manager struct {
- mh *indexedDbWorker.MessageHandler
+ mh *indexedDb2.MessageHandler
model dm.EventModel
}
diff --git a/indexedDb/dm/implementation.go b/indexedDb/dm/implementation.go
index 77b5fbc73d5ec3bea746597694342b522bd382de..7cd2f00e0bad630e3f331a1a6f66657cd6cdb77c 100644
--- a/indexedDb/dm/implementation.go
+++ b/indexedDb/dm/implementation.go
@@ -7,240 +7,375 @@
//go:build js && wasm
-package channelEventModel
+package main
import (
"crypto/ed25519"
"encoding/json"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
+ "strings"
+ "sync"
+ "syscall/js"
"time"
+ "github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/cmix/rounds"
"gitlab.com/elixxir/client/v4/dm"
+ "gitlab.com/elixxir/xxdk-wasm/utils"
+ "gitlab.com/xx_network/primitives/id"
+
+ "github.com/hack-pad/go-indexeddb/idb"
+ cryptoChannel "gitlab.com/elixxir/crypto/channel"
"gitlab.com/elixxir/crypto/message"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
)
// wasmModel implements dm.EventModel interface, which uses the channels system
// passed an object that adheres to in order to get events on the channel.
type wasmModel struct {
- wh *indexedDb.WorkerHandler
+ db *idb.Database
+ cipher cryptoChannel.Cipher
+ receivedMessageCB MessageReceivedCallback
+ updateMux sync.Mutex
+}
+
+// joinConversation is used for joining new conversations.
+func (w *wasmModel) joinConversation(nickname string,
+ pubKey ed25519.PublicKey, dmToken uint32, codeset uint8) error {
+ parentErr := errors.New("failed to joinConversation")
+
+ // Build object
+ newConvo := Conversation{
+ Pubkey: pubKey,
+ Nickname: nickname,
+ Token: dmToken,
+ CodesetVersion: codeset,
+ Blocked: false,
+ }
+
+ // Convert to jsObject
+ newConvoJson, err := json.Marshal(&newConvo)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to marshal Conversation: %+v", err)
+ }
+ convoObj, err := utils.JsonToJS(newConvoJson)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to marshal Conversation: %+v", err)
+ }
+
+ _, err = indexedDb2.Put(w.db, conversationStoreName, convoObj)
+ if err != nil {
+ return errors.WithMessagef(parentErr,
+ "Unable to put Conversation: %+v", err)
+ }
+ return nil
}
-// TransferMessage is JSON marshalled and sent to the worker.
-type TransferMessage struct {
- UUID uint64 `json:"uuid"`
- MessageID message.ID `json:"messageID"`
- ReactionTo message.ID `json:"reactionTo"`
- Nickname string `json:"nickname"`
- Text []byte `json:"text"`
- PubKey ed25519.PublicKey `json:"pubKey"`
- DmToken uint32 `json:"dmToken"`
- Codeset uint8 `json:"codeset"`
- Timestamp time.Time `json:"timestamp"`
- Round rounds.Round `json:"round"`
- MType dm.MessageType `json:"mType"`
- Status dm.Status `json:"status"`
+// buildMessage is a private helper that converts typical dm.EventModel inputs
+// into a basic Message structure for insertion into storage.
+//
+// NOTE: ID is not set inside this function because we want to use the
+// autoincrement key by default. If you are trying to overwrite an existing
+// message, then you need to set it manually yourself.
+func buildMessage(messageID, parentID, text []byte, pubKey ed25519.PublicKey,
+ timestamp time.Time, round id.Round, mType dm.MessageType,
+ status dm.Status) *Message {
+ return &Message{
+ MessageID: messageID,
+ ConversationPubKey: pubKey,
+ ParentMessageID: parentID,
+ Timestamp: timestamp,
+ Status: uint8(status),
+ Text: text,
+ Type: uint16(mType),
+ Round: uint64(round),
+ }
}
func (w *wasmModel) Receive(messageID message.ID, nickname string, text []byte,
pubKey ed25519.PublicKey, dmToken uint32, codeset uint8, timestamp time.Time,
round rounds.Round, mType dm.MessageType, status dm.Status) uint64 {
- msg := TransferMessage{
- MessageID: messageID,
- Nickname: nickname,
- Text: text,
- PubKey: pubKey,
- DmToken: dmToken,
- Codeset: codeset,
- Timestamp: timestamp,
- Round: round,
- MType: mType,
- Status: status,
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for TransferMessage: %+v", err)
+ parentErr := errors.New("failed to Receive")
+
+ // If there is no extant Conversation, create one.
+ _, err := indexedDb2.Get(
+ w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
+ if err != nil {
+ if strings.Contains(err.Error(), indexedDb2.ErrDoesNotExist) {
+ err = w.joinConversation(nickname, pubKey, dmToken, codeset)
+ if err != nil {
+ jww.ERROR.Printf("%+v", err)
+ }
+ } else {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get Conversation: %+v", err))
+ }
return 0
+ } else {
+ jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ if w.cipher != nil {
+ text, err = w.cipher.Encrypt(text)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to Receive: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
+ }
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about Receive", indexedDb.ResponseTimeout)
+ msgToInsert := buildMessage(messageID.Bytes(), nil, text, pubKey, timestamp,
+ round.ID, mType, status)
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive Message: %+v", err)
}
- return 0
+ go w.receivedMessageCB(uuid, pubKey, false)
+ return uuid
}
func (w *wasmModel) ReceiveText(messageID message.ID, nickname, text string,
pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- msg := TransferMessage{
- MessageID: messageID,
- Nickname: nickname,
- Text: []byte(text),
- PubKey: pubKey,
- DmToken: dmToken,
- Codeset: codeset,
- Timestamp: timestamp,
- Round: round,
- Status: status,
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for TransferMessage: %+v", err)
+ parentErr := errors.New("failed to ReceiveText")
+
+ // If there is no extant Conversation, create one.
+ _, err := indexedDb2.Get(
+ w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
+ if err != nil {
+ if strings.Contains(err.Error(), indexedDb2.ErrDoesNotExist) {
+ err = w.joinConversation(nickname, pubKey, dmToken, codeset)
+ if err != nil {
+ jww.ERROR.Printf("%+v", err)
+ }
+ } else {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get Conversation: %+v", err))
+ }
return 0
+ } else {
+ jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveTextTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ textBytes := []byte(text)
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt(textBytes)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveText: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
+ }
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveText", indexedDb.ResponseTimeout)
+ msgToInsert := buildMessage(messageID.Bytes(), nil, textBytes,
+ pubKey, timestamp, round.ID, dm.TextType, status)
+
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive Message: %+v", err)
}
- return 0
+ go w.receivedMessageCB(uuid, pubKey, false)
+ return uuid
}
func (w *wasmModel) ReceiveReply(messageID, reactionTo message.ID, nickname,
text string, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- msg := TransferMessage{
- MessageID: messageID,
- ReactionTo: reactionTo,
- Nickname: nickname,
- Text: []byte(text),
- PubKey: pubKey,
- DmToken: dmToken,
- Codeset: codeset,
- Timestamp: timestamp,
- Round: round,
- Status: status,
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for TransferMessage: %+v", err)
+ parentErr := errors.New("failed to ReceiveReply")
+
+ // If there is no extant Conversation, create one.
+ _, err := indexedDb2.Get(
+ w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
+ if err != nil {
+ if strings.Contains(err.Error(), indexedDb2.ErrDoesNotExist) {
+ err = w.joinConversation(nickname, pubKey, dmToken, codeset)
+ if err != nil {
+ jww.ERROR.Printf("%+v", err)
+ }
+ } else {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get Conversation: %+v", err))
+ }
return 0
+ } else {
+ jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveReplyTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ textBytes := []byte(text)
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt(textBytes)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveReply: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
+ }
+
+ msgToInsert := buildMessage(messageID.Bytes(), reactionTo.Marshal(), textBytes,
+ pubKey, timestamp, round.ID, dm.TextType, status)
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveReply", indexedDb.ResponseTimeout)
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive Message: %+v", err)
}
- return 0
+ go w.receivedMessageCB(uuid, pubKey, false)
+ return uuid
}
-func (w *wasmModel) ReceiveReaction(messageID, reactionTo message.ID, nickname,
+func (w *wasmModel) ReceiveReaction(messageID, _ message.ID, nickname,
reaction string, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- msg := TransferMessage{
- MessageID: messageID,
- ReactionTo: reactionTo,
- Nickname: nickname,
- Text: []byte(reaction),
- PubKey: pubKey,
- DmToken: dmToken,
- Codeset: codeset,
- Timestamp: timestamp,
- Round: round,
- Status: status,
- }
-
- data, err := json.Marshal(msg)
- if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for TransferMessage: %+v", err)
+ parentErr := errors.New("failed to ReceiveText")
+
+ // If there is no extant Conversation, create one.
+ _, err := indexedDb2.Get(
+ w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
+ if err != nil {
+ if strings.Contains(err.Error(), indexedDb2.ErrDoesNotExist) {
+ err = w.joinConversation(nickname, pubKey, dmToken, codeset)
+ if err != nil {
+ jww.ERROR.Printf("%+v", err)
+ }
+ } else {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get Conversation: %+v", err))
+ }
return 0
+ } else {
+ jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- uuidChan := make(chan uint64)
- w.wh.SendMessage(indexedDb.ReceiveReactionTag, data, func(data []byte) {
- var uuid uint64
- err = json.Unmarshal(data, &uuid)
+ // Handle encryption, if it is present
+ textBytes := []byte(reaction)
+ if w.cipher != nil {
+ textBytes, err = w.cipher.Encrypt(textBytes)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON unmarshal response to ReceiveReaction: %+v", err)
- uuidChan <- 0
+ jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
+ return 0
}
- uuidChan <- uuid
- })
+ }
+
+ msgToInsert := buildMessage(messageID.Bytes(), nil, textBytes,
+ pubKey, timestamp, round.ID, dm.ReactionType, status)
- select {
- case uuid := <-uuidChan:
- return uuid
- case <-time.After(indexedDb.ResponseTimeout):
- jww.ERROR.Printf("Timed out after %s waiting for response from the "+
- "worker about ReceiveReaction", indexedDb.ResponseTimeout)
+ uuid, err := w.receiveHelper(msgToInsert, false)
+ if err != nil {
+ jww.ERROR.Printf("Failed to receive Message: %+v", err)
}
- return 0
+ go w.receivedMessageCB(uuid, pubKey, false)
+ return uuid
}
func (w *wasmModel) UpdateSentStatus(uuid uint64, messageID message.ID,
timestamp time.Time, round rounds.Round, status dm.Status) {
- msg := TransferMessage{
- UUID: uuid,
- MessageID: messageID,
- Timestamp: timestamp,
- Round: round,
- Status: status,
+ parentErr := errors.New("failed to UpdateSentStatus")
+
+ // FIXME: this is a bit of race condition without the mux.
+ // This should be done via the transactions (i.e., make a
+ // special version of receiveHelper)
+ w.updateMux.Lock()
+ defer w.updateMux.Unlock()
+
+ // Convert messageID to the key generated by json.Marshal
+ key := js.ValueOf(uuid)
+
+ // Use the key to get the existing Message
+ currentMsg, err := indexedDb2.Get(w.db, messageStoreName, key)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Unable to get message: %+v", err))
+ return
+ }
+
+ // Extract the existing Message and update the Status
+ newMessage := &Message{}
+ err = json.Unmarshal([]byte(utils.JsToJson(currentMsg)), newMessage)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
+ "Could not JSON unmarshal message: %+v", err))
+ return
+ }
+
+ newMessage.Status = uint8(status)
+ if !messageID.Equals(message.ID{}) {
+ newMessage.MessageID = messageID.Bytes()
+ }
+
+ if round.ID != 0 {
+ newMessage.Round = uint64(round.ID)
+ }
+
+ if !timestamp.Equal(time.Time{}) {
+ newMessage.Timestamp = timestamp
+ }
+
+ // Store the updated Message
+ _, err = w.receiveHelper(newMessage, true)
+ if err != nil {
+ jww.ERROR.Printf("%+v", errors.Wrap(parentErr, err.Error()))
+ }
+ go w.receivedMessageCB(uuid, newMessage.ConversationPubKey, true)
+}
+
+// receiveHelper is a private helper for receiving any sort of message.
+func (w *wasmModel) receiveHelper(
+ newMessage *Message, isUpdate bool) (uint64, error) {
+ // Convert to jsObject
+ newMessageJson, err := json.Marshal(newMessage)
+ if err != nil {
+ return 0, errors.Errorf("Unable to marshal Message: %+v", err)
+ }
+ messageObj, err := utils.JsonToJS(newMessageJson)
+ if err != nil {
+ return 0, errors.Errorf("Unable to marshal Message: %+v", err)
+ }
+
+ // Unset the primaryKey for inserts so that it can be auto-populated and
+ // incremented
+ if !isUpdate {
+ messageObj.Delete("id")
}
- data, err := json.Marshal(msg)
+ // Store message to database
+ result, err := indexedDb2.Put(w.db, messageStoreName, messageObj)
if err != nil {
- jww.ERROR.Printf(
- "Could not JSON marshal payload for TransferMessage: %+v", err)
+ return 0, errors.Errorf("Unable to put Message: %+v", err)
}
- w.wh.SendMessage(indexedDb.UpdateSentStatusTag, data, nil)
+ // NOTE: Sometimes the insert fails to return an error but hits a duplicate
+ // insert, so this fallthrough returns the UUID entry in that case.
+ if result.IsUndefined() {
+ msgID := message.ID{}
+ copy(msgID[:], newMessage.MessageID)
+ uuid, errLookup := w.msgIDLookup(msgID)
+ if uuid != 0 && errLookup == nil {
+ return uuid, nil
+ }
+ return 0, errors.Errorf("uuid lookup failure: %+v", err)
+ }
+ uuid := uint64(result.Int())
+ jww.DEBUG.Printf("Successfully stored message %d", uuid)
+
+ return uuid, nil
+}
+
+// msgIDLookup gets the UUID of the Message with the given messageID.
+func (w *wasmModel) msgIDLookup(messageID message.ID) (uint64, error) {
+ resultObj, err := indexedDb2.GetIndex(w.db, messageStoreName,
+ messageStoreMessageIndex, utils.CopyBytesToJS(messageID.Marshal()))
+ if err != nil {
+ return 0, err
+ }
+
+ uuid := uint64(0)
+ if !resultObj.IsUndefined() {
+ uuid = uint64(resultObj.Get("id").Int())
+ }
+ return uuid, nil
}
diff --git a/indexedDb/dm/init.go b/indexedDb/dm/init.go
index f3bc313ed97dfd0755b36f25a7adc6bd74e746aa..66c9c34a68d8940faf14aa87a1f3b2ba2e020cc9 100644
--- a/indexedDb/dm/init.go
+++ b/indexedDb/dm/init.go
@@ -7,24 +7,31 @@
//go:build js && wasm
-package channelEventModel
+package main
import (
"crypto/ed25519"
- "encoding/json"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
+ "syscall/js"
"time"
+ "github.com/hack-pad/go-indexeddb/idb"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/dm"
cryptoChannel "gitlab.com/elixxir/crypto/channel"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
- "gitlab.com/elixxir/xxdk-wasm/storage"
)
-// WorkerJavascriptFileURL is the URL of the script the worker will execute to
-// launch the worker WASM binary. It must obey the same-origin policy.
-const WorkerJavascriptFileURL = "/integrations/assets/indexedDbWorker.js"
+const (
+ // databaseSuffix is the suffix to be appended to the name of
+ // the database.
+ databaseSuffix = "_speakeasy_dm"
+
+ // currentVersion is the current version of the IndexDb
+ // runtime. Used for migration purposes.
+ currentVersion uint = 1
+)
// MessageReceivedCallback is called any time a message is received or updated.
//
@@ -32,134 +39,192 @@ const WorkerJavascriptFileURL = "/integrations/assets/indexedDbWorker.js"
type MessageReceivedCallback func(
uuid uint64, pubKey ed25519.PublicKey, update bool)
-// NewWASMEventModelMessage is JSON marshalled and sent to the worker for
-// [NewWASMEventModel].
-type NewWASMEventModelMessage struct {
- Path string `json:"path"`
- EncryptionJSON string `json:"encryptionJSON"`
-}
-
// NewWASMEventModel returns a [channels.EventModel] backed by a wasmModel.
// The name should be a base64 encoding of the users public key.
func NewWASMEventModel(path string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback) (dm.EventModel, error) {
+ cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
+ dm.EventModel, error) {
+ databaseName := path + databaseSuffix
+ return newWASMModel(databaseName, encryption, cb, storeEncryptionStatus)
+}
- // TODO: bring in URL and name from caller
- wh, err := indexedDb.NewWorkerHandler(
- WorkerJavascriptFileURL, "dmIndexedDb")
+// storeEncryptionStatusFn matches storage.StoreIndexedDbEncryptionStatus so
+// that the data can be sent between the worker and main thread.
+type storeEncryptionStatusFn func(
+ databaseName string, encryptionStatus bool) (bool, error)
+
+// newWASMModel creates the given [idb.Database] and returns a wasmModel.
+func newWASMModel(databaseName string, encryption cryptoChannel.Cipher,
+ cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
+ *wasmModel, error) {
+ // Attempt to open database object
+ ctx, cancel := indexedDb2.NewContext()
+ defer cancel()
+ openRequest, err := idb.Global().Open(ctx, databaseName, currentVersion,
+ func(db *idb.Database, oldVersion, newVersion uint) error {
+ if oldVersion == newVersion {
+ jww.INFO.Printf("IndexDb version is current: v%d",
+ newVersion)
+ return nil
+ }
+
+ jww.INFO.Printf("IndexDb upgrade required: v%d -> v%d",
+ oldVersion, newVersion)
+
+ if oldVersion == 0 && newVersion >= 1 {
+ err := v1Upgrade(db)
+ if err != nil {
+ return err
+ }
+ oldVersion = 1
+ }
+
+ // if oldVersion == 1 && newVersion >= 2 { v2Upgrade(), oldVersion = 2 }
+ return nil
+ })
if err != nil {
return nil, err
}
- // Register handler to manage messages for the MessageReceivedCallback
- wh.RegisterHandler(indexedDb.GetMessageTag, indexedDb.InitID, false,
- messageReceivedCallbackHandler(cb))
-
- // Register handler to manage checking encryption status from local storage
- wh.RegisterHandler(indexedDb.EncryptionStatusTag, indexedDb.InitID, false,
- checkDbEncryptionStatusHandler(wh))
+ // Wait for database open to finish
+ db, err := openRequest.Await(ctx)
+ if err != nil {
+ return nil, err
+ }
- encryptionJSON, err := json.Marshal(encryption)
+ // Save the encryption status to storage
+ encryptionStatus := encryption != nil
+ loadedEncryptionStatus, err :=
+ storeEncryptionStatus(databaseName, encryptionStatus)
if err != nil {
return nil, err
}
- message := NewWASMEventModelMessage{
- Path: path,
- EncryptionJSON: string(encryptionJSON),
+ // Verify encryption status does not change
+ if encryptionStatus != loadedEncryptionStatus {
+ return nil, errors.New(
+ "Cannot load database with different encryption status.")
+ } else if !encryptionStatus {
+ jww.WARN.Printf("IndexedDb encryption disabled!")
}
- payload, err := json.Marshal(message)
+ // Attempt to ensure the database has been properly initialized
+ openRequest, err = idb.Global().Open(ctx, databaseName, currentVersion,
+ func(db *idb.Database, oldVersion, newVersion uint) error {
+ return nil
+ })
if err != nil {
return nil, err
}
-
- errChan := make(chan string)
- wh.SendMessage(indexedDb.NewWASMEventModelTag, payload, func(data []byte) {
- errChan <- string(data)
- })
-
- select {
- case workerErr := <-errChan:
- if workerErr != "" {
- return nil, errors.New(workerErr)
- }
- case <-time.After(indexedDb.ResponseTimeout):
- return nil, errors.Errorf("timed out after %s waiting for indexedDB "+
- "database in worker to intialize", indexedDb.ResponseTimeout)
+ // Wait for database open to finish
+ db, err = openRequest.Await(ctx)
+ if err != nil {
+ return nil, err
}
+ wrapper := &wasmModel{db: db, receivedMessageCB: cb, cipher: encryption}
- return &wasmModel{wh}, nil
+ return wrapper, nil
}
-// MessageReceivedCallbackMessage is JSON marshalled and received from the
-// worker for the [MessageReceivedCallback] callback.
-type MessageReceivedCallbackMessage struct {
- UUID uint64 `json:"uuid"`
- PubKey ed25519.PublicKey `json:"pubKey"`
- Update bool `json:"update"`
-}
+// v1Upgrade performs the v0 -> v1 database upgrade.
+//
+// This can never be changed without permanently breaking backwards
+// compatibility.
+func v1Upgrade(db *idb.Database) error {
+ indexOpts := idb.IndexOptions{
+ Unique: false,
+ MultiEntry: false,
+ }
-// messageReceivedCallbackHandler returns a handler to manage messages for the
-// MessageReceivedCallback.
-func messageReceivedCallbackHandler(cb MessageReceivedCallback) func(data []byte) {
- return func(data []byte) {
- var msg MessageReceivedCallbackMessage
- err := json.Unmarshal(data, &msg)
- if err != nil {
- jww.ERROR.Printf("Failed to JSON unmarshal "+
- "MessageReceivedCallback message from worker: %+v", err)
- return
- }
- cb(msg.UUID, msg.PubKey, msg.Update)
+ // Build Message ObjectStore and Indexes
+ messageStoreOpts := idb.ObjectStoreOptions{
+ KeyPath: js.ValueOf(msgPkeyName),
+ AutoIncrement: true,
+ }
+ messageStore, err := db.CreateObjectStore(messageStoreName, messageStoreOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreMessageIndex,
+ js.ValueOf(messageStoreMessage),
+ idb.IndexOptions{
+ Unique: true,
+ MultiEntry: false,
+ })
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreConversationIndex,
+ js.ValueOf(messageStoreConversation), indexOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreParentIndex,
+ js.ValueOf(messageStoreParent), indexOpts)
+ if err != nil {
+ return err
+ }
+ _, err = messageStore.CreateIndex(messageStoreTimestampIndex,
+ js.ValueOf(messageStoreTimestamp), indexOpts)
+ if err != nil {
+ return err
}
-}
-// EncryptionStatusMessage is JSON marshalled and received from the worker when
-// the database checks if it is encrypted.
-type EncryptionStatusMessage struct {
- DatabaseName string `json:"databaseName"`
- EncryptionStatus bool `json:"encryptionStatus"`
-}
+ // Build Channel ObjectStore
+ conversationStoreOpts := idb.ObjectStoreOptions{
+ KeyPath: js.ValueOf(convoPkeyName),
+ AutoIncrement: false,
+ }
+ _, err = db.CreateObjectStore(conversationStoreName, conversationStoreOpts)
+ if err != nil {
+ return err
+ }
-// EncryptionStatusReply is JSON marshalled and sent to the worker is response
-// to the [EncryptionStatusMessage].
-type EncryptionStatusReply struct {
- EncryptionStatus bool `json:"encryptionStatus"`
- Error string `json:"error"`
-}
+ // Get the database name and save it to storage
+ if databaseName, err2 := db.Name(); err2 != nil {
+ return err2
+ } else if err = storeDatabaseName(databaseName); err != nil {
+ return err
+ }
-// checkDbEncryptionStatusHandler returns a handler to manage checking
-// encryption status from local storage.
-func checkDbEncryptionStatusHandler(wh *indexedDb.WorkerHandler) func(data []byte) {
- return func(data []byte) {
- // Unmarshal received message
- var msg EncryptionStatusMessage
- err := json.Unmarshal(data, &msg)
- if err != nil {
- jww.ERROR.Printf("Failed to JSON unmarshal "+
- "EncryptionStatusMessage message from worker: %+v", err)
- return
- }
+ return nil
+}
- // Pass message values to storage
- loadedEncryptionStatus, err := storage.StoreIndexedDbEncryptionStatus(
- msg.DatabaseName, msg.EncryptionStatus)
- var reply EncryptionStatusReply
- if err != nil {
- reply.Error = err.Error()
- } else {
- reply.EncryptionStatus = loadedEncryptionStatus
- }
+// storeDatabaseName sends the database name to storage.StoreIndexedDb in the
+// main thread to be stored in localstorage and waits for the error to be
+// returned.
+//
+// The function specified below is a placeholder until set by
+// registerDatabaseNameStore. registerDatabaseNameStore must be called before
+// storeDatabaseName.
+var storeDatabaseName = func(databaseName string) error { return nil }
+
+// RegisterDatabaseNameStore sets storeDatabaseName to send the database to
+// storage.StoreIndexedDb in the main thread when called and registers a handler
+// to listen for the response.
+func RegisterDatabaseNameStore(m *manager) {
+ storeDatabaseNameResponseChan := make(chan []byte)
+ // Register handler
+ m.mh.RegisterHandler(indexedDb.StoreDatabaseNameTag, func(data []byte) []byte {
+ storeDatabaseNameResponseChan <- data
+ return nil
+ })
- // Return response
- statusData, err := json.Marshal(reply)
- if err != nil {
- jww.ERROR.Printf(
- "Failed to JSON marshal EncryptionStatusReply: %+v", err)
- return
+ storeDatabaseName = func(databaseName string) error {
+ m.mh.SendResponse(indexedDb.StoreDatabaseNameTag, indexedDb.InitID,
+ []byte(databaseName))
+
+ // Wait for response
+ select {
+ case response := <-storeDatabaseNameResponseChan:
+ if len(response) > 0 {
+ return errors.New(string(response))
+ }
+ case <-time.After(indexedDb.ResponseTimeout):
+ return errors.Errorf("timed out after %s waiting for "+
+ "response about storing the database name in local "+
+ "storage in the main thread", indexedDb.ResponseTimeout)
}
-
- wh.SendMessage(indexedDb.EncryptionStatusTag, statusData, nil)
+ return nil
}
}
diff --git a/indexedDbWorker/dm/main.go b/indexedDb/dm/main.go
similarity index 87%
rename from indexedDbWorker/dm/main.go
rename to indexedDb/dm/main.go
index d720ddee52b525d98d1f6027927dd8b750182054..5f0d16dc868a7f8fd1bf67158ec3b9f8e5ffbd90 100644
--- a/indexedDbWorker/dm/main.go
+++ b/indexedDb/dm/main.go
@@ -11,13 +11,13 @@ package main
import (
"fmt"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDb"
)
func main() {
fmt.Println("Starting xxDK WebAssembly Database Worker.")
- m := &manager{mh: indexedDbWorker.NewMessageHandler()}
+ m := &manager{mh: indexedDb2.NewMessageHandler()}
m.RegisterHandlers()
RegisterDatabaseNameStore(m)
m.mh.SignalReady()
diff --git a/indexedDbWorker/dm/model.go b/indexedDb/dm/model.go
similarity index 100%
rename from indexedDbWorker/dm/model.go
rename to indexedDb/dm/model.go
diff --git a/indexedDbWorker/messageHandler.go b/indexedDb/messageHandler.go
similarity index 98%
rename from indexedDbWorker/messageHandler.go
rename to indexedDb/messageHandler.go
index da05af24cfca4403926f627bf0c54d951eb216a0..ad729ee8cf9c43116903f5cb4727e48724b78a56 100644
--- a/indexedDbWorker/messageHandler.go
+++ b/indexedDb/messageHandler.go
@@ -7,13 +7,13 @@
//go:build js && wasm
-package indexedDbWorker
+package indexedDb2
import (
"encoding/json"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
"gitlab.com/elixxir/xxdk-wasm/utils"
"sync"
"syscall/js"
diff --git a/indexedDbWorker/utils.go b/indexedDb/utils.go
similarity index 99%
rename from indexedDbWorker/utils.go
rename to indexedDb/utils.go
index 321dd082214e7ab88ed6351442882317426f4e87..87696d0f8be15f8584d7c1dbf45a215767e63467 100644
--- a/indexedDbWorker/utils.go
+++ b/indexedDb/utils.go
@@ -10,7 +10,7 @@
// This file contains several generic IndexedDB helper functions that
// may be useful for any IndexedDB implementations.
-package indexedDbWorker
+package indexedDb2
import (
"context"
@@ -31,7 +31,7 @@ const (
ErrDoesNotExist = "result is undefined"
)
-// NewContext builds a context for indexedDb operations.
+// NewContext builds a context for indexedDbWorker operations.
func NewContext() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), dbTimeout)
}
diff --git a/indexedDbWorker/utils_test.go b/indexedDb/utils_test.go
similarity index 98%
rename from indexedDbWorker/utils_test.go
rename to indexedDb/utils_test.go
index 62851adb196907b392a85d82b5e333990cc26122..279a33f9f86cfcaed014a9fe0c64db34d4a5cbac 100644
--- a/indexedDbWorker/utils_test.go
+++ b/indexedDb/utils_test.go
@@ -7,7 +7,7 @@
//go:build js && wasm
-package indexedDbWorker
+package indexedDb2
import (
"github.com/hack-pad/go-indexeddb/idb"
diff --git a/indexedDbWorker/channels/implementation.go b/indexedDbWorker/channels/implementation.go
index 96e4351964a024155041a92892d2b6328c932c4e..fd7e13000c496a25d8bb1b226efa9490e526ee55 100644
--- a/indexedDbWorker/channels/implementation.go
+++ b/indexedDbWorker/channels/implementation.go
@@ -7,28 +7,22 @@
//go:build js && wasm
-package main
+package channels
import (
"crypto/ed25519"
- "encoding/base64"
"encoding/json"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
- "strings"
- "sync"
- "syscall/js"
"time"
- "github.com/hack-pad/go-indexeddb/idb"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/channels"
"gitlab.com/elixxir/client/v4/cmix/rounds"
cryptoBroadcast "gitlab.com/elixxir/crypto/broadcast"
- cryptoChannel "gitlab.com/elixxir/crypto/channel"
"gitlab.com/elixxir/crypto/message"
- "gitlab.com/elixxir/xxdk-wasm/utils"
"gitlab.com/xx_network/primitives/id"
)
@@ -36,108 +30,23 @@ import (
// system passed an object that adheres to in order to get events on the
// channel.
type wasmModel struct {
- db *idb.Database
- cipher cryptoChannel.Cipher
- receivedMessageCB MessageReceivedCallback
- updateMux sync.Mutex
+ wh *indexedDb.WorkerHandler
}
// JoinChannel is called whenever a channel is joined locally.
func (w *wasmModel) JoinChannel(channel *cryptoBroadcast.Channel) {
- parentErr := errors.New("failed to JoinChannel")
-
- // Build object
- newChannel := Channel{
- ID: channel.ReceptionID.Marshal(),
- Name: channel.Name,
- Description: channel.Description,
- }
-
- // Convert to jsObject
- newChannelJson, err := json.Marshal(&newChannel)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to marshal Channel: %+v", err))
- return
- }
- channelObj, err := utils.JsonToJS(newChannelJson)
+ data, err := json.Marshal(channel)
if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to marshal Channel: %+v", err))
+ jww.ERROR.Printf("Could not JSON marshal broadcast.Channel: %+v", err)
return
}
- _, err = indexedDbWorker.Put(w.db, channelsStoreName, channelObj)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to put Channel: %+v", err))
- }
+ w.wh.SendMessage(indexedDb.JoinChannelTag, data, nil)
}
// LeaveChannel is called whenever a channel is left locally.
func (w *wasmModel) LeaveChannel(channelID *id.ID) {
- parentErr := errors.New("failed to LeaveChannel")
-
- // Delete the channel from storage
- err := indexedDbWorker.Delete(w.db, channelsStoreName,
- js.ValueOf(channelID.String()))
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to delete Channel: %+v", err))
- return
- }
-
- // Clean up lingering data
- err = w.deleteMsgByChannel(channelID)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Deleting Channel's Message data failed: %+v", err))
- return
- }
- jww.DEBUG.Printf("Successfully deleted channel: %s", channelID)
-}
-
-// deleteMsgByChannel is a private helper that uses messageStoreChannelIndex
-// to delete all Message with the given Channel ID.
-func (w *wasmModel) deleteMsgByChannel(channelID *id.ID) error {
- parentErr := errors.New("failed to deleteMsgByChannel")
-
- // Prepare the Transaction
- txn, err := w.db.Transaction(idb.TransactionReadWrite, messageStoreName)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to create Transaction: %+v", err)
- }
- store, err := txn.ObjectStore(messageStoreName)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to get ObjectStore: %+v", err)
- }
- index, err := store.Index(messageStoreChannelIndex)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to get Index: %+v", err)
- }
-
- // Perform the operation
- channelIdStr := base64.StdEncoding.EncodeToString(channelID.Marshal())
- keyRange, err := idb.NewKeyRangeOnly(js.ValueOf(channelIdStr))
- cursorRequest, err := index.OpenCursorRange(keyRange, idb.CursorNext)
- if err != nil {
- return errors.WithMessagef(parentErr, "Unable to open Cursor: %+v", err)
- }
- ctx, cancel := indexedDbWorker.NewContext()
- err = cursorRequest.Iter(ctx,
- func(cursor *idb.CursorWithValue) error {
- _, err := cursor.Delete()
- return err
- })
- cancel()
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to delete Message data: %+v", err)
- }
- return nil
+ w.wh.SendMessage(indexedDb.LeaveChannelTag, channelID.Marshal(), nil)
}
// ReceiveMessage is called whenever a message is received on a given channel.
@@ -148,30 +57,58 @@ func (w *wasmModel) ReceiveMessage(channelID *id.ID, messageID message.ID,
nickname, text string, pubKey ed25519.PublicKey, dmToken uint32,
codeset uint8, timestamp time.Time, lease time.Duration, round rounds.Round,
mType channels.MessageType, status channels.SentStatus, hidden bool) uint64 {
- textBytes := []byte(text)
- var err error
+ msg := channels.ModelMessage{
+ Nickname: nickname,
+ MessageID: messageID,
+ ChannelID: channelID,
+ Timestamp: timestamp,
+ Lease: lease,
+ Status: status,
+ Hidden: hidden,
+ Pinned: false,
+ Content: []byte(text),
+ Type: mType,
+ Round: round.ID,
+ PubKey: pubKey,
+ CodesetVersion: codeset,
+ DmToken: dmToken,
+ }
- // Handle encryption, if it is present
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt([]byte(text))
- if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
- }
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for ReceiveMessage: %+v", err)
+ return 0
}
- msgToInsert := buildMessage(
- channelID.Marshal(), messageID.Bytes(), nil, nickname,
- textBytes, pubKey, dmToken, codeset, timestamp, lease, round.ID, mType,
- false, hidden, status)
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveMessageTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveMessage: %+v", err)
+ uuidChan <- 0
+ }
+ uuidChan <- uuid
+ })
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveMessage", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, channelID, false)
- return uuid
+ return 0
+}
+
+// ReceiveReplyMessage is JSON marshalled and sent to the worker for
+// [wasmModel.ReceiveReply].
+type ReceiveReplyMessage struct {
+ ReactionTo message.ID `json:"replyTo"`
+ channels.ModelMessage `json:"message"`
}
// ReceiveReply is called whenever a message is received that is a reply on a
@@ -185,29 +122,54 @@ func (w *wasmModel) ReceiveReply(channelID *id.ID, messageID,
dmToken uint32, codeset uint8, timestamp time.Time, lease time.Duration,
round rounds.Round, mType channels.MessageType, status channels.SentStatus,
hidden bool) uint64 {
- textBytes := []byte(text)
- var err error
+ msg := ReceiveReplyMessage{
+ ReactionTo: replyTo,
+ ModelMessage: channels.ModelMessage{
+ Nickname: nickname,
+ MessageID: messageID,
+ ChannelID: channelID,
+ Timestamp: timestamp,
+ Lease: lease,
+ Status: status,
+ Hidden: hidden,
+ Pinned: false,
+ Content: []byte(text),
+ Type: mType,
+ Round: round.ID,
+ PubKey: pubKey,
+ CodesetVersion: codeset,
+ DmToken: dmToken,
+ },
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for ReceiveReply: %+v", err)
+ return 0
+ }
- // Handle encryption, if it is present
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt([]byte(text))
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveReplyTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveReply: %+v", err)
+ uuidChan <- 0
}
- }
+ uuidChan <- uuid
+ })
- msgToInsert := buildMessage(channelID.Marshal(), messageID.Bytes(),
- replyTo.Bytes(), nickname, textBytes, pubKey, dmToken, codeset,
- timestamp, lease, round.ID, mType, hidden, false, status)
-
- uuid, err := w.receiveHelper(msgToInsert, false)
-
- if err != nil {
- jww.ERROR.Printf("Failed to receive reply: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, channelID, false)
- return uuid
+
+ return 0
}
// ReceiveReaction is called whenever a reaction to a message is received on a
@@ -221,29 +183,79 @@ func (w *wasmModel) ReceiveReaction(channelID *id.ID, messageID,
dmToken uint32, codeset uint8, timestamp time.Time, lease time.Duration,
round rounds.Round, mType channels.MessageType, status channels.SentStatus,
hidden bool) uint64 {
- textBytes := []byte(reaction)
- var err error
- // Handle encryption, if it is present
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt([]byte(reaction))
+ msg := ReceiveReplyMessage{
+ ReactionTo: reactionTo,
+ ModelMessage: channels.ModelMessage{
+ Nickname: nickname,
+ MessageID: messageID,
+ ChannelID: channelID,
+ Timestamp: timestamp,
+ Lease: lease,
+ Status: status,
+ Hidden: hidden,
+ Pinned: false,
+ Content: []byte(reaction),
+ Type: mType,
+ Round: round.ID,
+ PubKey: pubKey,
+ CodesetVersion: codeset,
+ DmToken: dmToken,
+ },
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for ReceiveReaction: %+v", err)
+ return 0
+ }
+
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveReactionTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveReaction: %+v", err)
+ uuidChan <- 0
}
+ uuidChan <- uuid
+ })
+
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- msgToInsert := buildMessage(
- channelID.Marshal(), messageID.Bytes(), reactionTo.Bytes(), nickname,
- textBytes, pubKey, dmToken, codeset, timestamp, lease, round.ID, mType,
- false, hidden, status)
+ return 0
+}
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive reaction: %+v", err)
- }
- go w.receivedMessageCB(uuid, channelID, false)
- return uuid
+// MessageUpdateInfo is JSON marshalled and sent to the worker for
+// [wasmModel.UpdateFromMessageID] and [wasmModel.UpdateFromUUID].
+type MessageUpdateInfo struct {
+ UUID uint64 `json:"uuid"`
+
+ MessageID message.ID `json:"messageID"`
+ MessageIDSet bool `json:"messageIDSet"`
+
+ Timestamp time.Time `json:"timestamp"`
+ TimestampSet bool `json:"timestampSet"`
+
+ RoundID id.Round `json:"round"`
+ RoundIDSet bool `json:"roundIDSet"`
+
+ Pinned bool `json:"pinned"`
+ PinnedSet bool `json:"pinnedSet"`
+
+ Hidden bool `json:"hidden"`
+ HiddenSet bool `json:"hiddenSet"`
+
+ Status channels.SentStatus `json:"status"`
+ StatusSet bool `json:"statusSet"`
}
// UpdateFromUUID is called whenever a message at the UUID is modified.
@@ -254,31 +266,40 @@ func (w *wasmModel) ReceiveReaction(channelID *id.ID, messageID,
func (w *wasmModel) UpdateFromUUID(uuid uint64, messageID *message.ID,
timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
status *channels.SentStatus) {
- parentErr := errors.New("failed to UpdateFromUUID")
-
- // FIXME: this is a bit of race condition without the mux.
- // This should be done via the transactions (i.e., make a
- // special version of receiveHelper)
- w.updateMux.Lock()
- defer w.updateMux.Unlock()
-
- // Convert messageID to the key generated by json.Marshal
- key := js.ValueOf(uuid)
+ msg := MessageUpdateInfo{UUID: uuid}
+ if messageID != nil {
+ msg.MessageID = *messageID
+ msg.MessageIDSet = true
+ }
+ if timestamp != nil {
+ msg.Timestamp = *timestamp
+ msg.TimestampSet = true
+ }
+ if round != nil {
+ msg.RoundID = round.ID
+ msg.RoundIDSet = true
+ }
+ if pinned != nil {
+ msg.Pinned = *pinned
+ msg.PinnedSet = true
+ }
+ if hidden != nil {
+ msg.Hidden = *hidden
+ msg.HiddenSet = true
+ }
+ if status != nil {
+ msg.Status = *status
+ msg.StatusSet = true
+ }
- // Use the key to get the existing Message
- currentMsg, err := indexedDbWorker.Get(w.db, messageStoreName, key)
+ data, err := json.Marshal(msg)
if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get message: %+v", err))
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for UpdateFromUUID: %+v", err)
return
}
- _, err = w.updateMessage(utils.JsToJson(currentMsg), messageID, timestamp,
- round, pinned, hidden, status)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to updateMessage: %+v", err))
- }
+ w.wh.SendMessage(indexedDb.UpdateFromUUIDTag, data, nil)
}
// UpdateFromMessageID is called whenever a message with the message ID is
@@ -293,222 +314,109 @@ func (w *wasmModel) UpdateFromUUID(uuid uint64, messageID *message.ID,
func (w *wasmModel) UpdateFromMessageID(messageID message.ID,
timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
status *channels.SentStatus) uint64 {
- parentErr := errors.New("failed to UpdateFromMessageID")
-
- // FIXME: this is a bit of race condition without the mux.
- // This should be done via the transactions (i.e., make a
- // special version of receiveHelper)
- w.updateMux.Lock()
- defer w.updateMux.Unlock()
-
- msgIDStr := base64.StdEncoding.EncodeToString(messageID.Marshal())
- currentMsgObj, err := indexedDbWorker.GetIndex(w.db, messageStoreName,
- messageStoreMessageIndex, js.ValueOf(msgIDStr))
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Failed to get message by index: %+v", err))
- return 0
- }
- currentMsg := utils.JsToJson(currentMsgObj)
- uuid, err := w.updateMessage(currentMsg, &messageID, timestamp,
- round, pinned, hidden, status)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to updateMessage: %+v", err))
- }
- return uuid
-}
-
-// updateMessage is a helper for updating a stored message.
-func (w *wasmModel) updateMessage(currentMsgJson string, messageID *message.ID,
- timestamp *time.Time, round *rounds.Round, pinned, hidden *bool,
- status *channels.SentStatus) (uint64, error) {
-
- newMessage := &Message{}
- err := json.Unmarshal([]byte(currentMsgJson), newMessage)
- if err != nil {
- return 0, err
- }
-
- if status != nil {
- newMessage.Status = uint8(*status)
- }
- if messageID != nil {
- newMessage.MessageID = messageID.Bytes()
+ msg := MessageUpdateInfo{MessageID: messageID, MessageIDSet: true}
+ if timestamp != nil {
+ msg.Timestamp = *timestamp
+ msg.TimestampSet = true
}
-
if round != nil {
- newMessage.Round = uint64(round.ID)
- }
-
- if timestamp != nil {
- newMessage.Timestamp = *timestamp
+ msg.RoundID = round.ID
+ msg.RoundIDSet = true
}
-
if pinned != nil {
- newMessage.Pinned = *pinned
+ msg.Pinned = *pinned
+ msg.PinnedSet = true
}
-
if hidden != nil {
- newMessage.Hidden = *hidden
- }
-
- // Store the updated Message
- uuid, err := w.receiveHelper(newMessage, true)
- if err != nil {
- return 0, err
+ msg.Hidden = *hidden
+ msg.HiddenSet = true
}
- channelID := &id.ID{}
- copy(channelID[:], newMessage.ChannelID)
- go w.receivedMessageCB(uuid, channelID, true)
-
- return uuid, nil
-}
-
-// buildMessage is a private helper that converts typical [channels.EventModel]
-// inputs into a basic Message structure for insertion into storage.
-//
-// NOTE: ID is not set inside this function because we want to use the
-// autoincrement key by default. If you are trying to overwrite an existing
-// message, then you need to set it manually yourself.
-func buildMessage(channelID, messageID, parentID []byte, nickname string,
- text []byte, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
- timestamp time.Time, lease time.Duration, round id.Round,
- mType channels.MessageType, pinned, hidden bool,
- status channels.SentStatus) *Message {
- return &Message{
- MessageID: messageID,
- Nickname: nickname,
- ChannelID: channelID,
- ParentMessageID: parentID,
- Timestamp: timestamp,
- Lease: lease,
- Status: uint8(status),
- Hidden: hidden,
- Pinned: pinned,
- Text: text,
- Type: uint16(mType),
- Round: uint64(round),
- // User Identity Info
- Pubkey: pubKey,
- DmToken: dmToken,
- CodesetVersion: codeset,
+ if status != nil {
+ msg.Status = *status
+ msg.StatusSet = true
}
-}
-// receiveHelper is a private helper for receiving any sort of message.
-func (w *wasmModel) receiveHelper(
- newMessage *Message, isUpdate bool) (uint64, error) {
- // Convert to jsObject
- newMessageJson, err := json.Marshal(newMessage)
+ data, err := json.Marshal(msg)
if err != nil {
- return 0, errors.Errorf("Unable to marshal Message: %+v", err)
- }
- messageObj, err := utils.JsonToJS(newMessageJson)
- if err != nil {
- return 0, errors.Errorf("Unable to marshal Message: %+v", err)
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for UpdateFromMessageID: %+v", err)
+ return 0
}
- // Unset the primaryKey for inserts so that it can be auto-populated and
- // incremented
- if !isUpdate {
- messageObj.Delete("id")
- }
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.UpdateFromMessageIDTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to UpdateFromMessageID: %+v", err)
+ uuidChan <- 0
+ }
+ uuidChan <- uuid
+ })
- // Store message to database
- result, err := indexedDbWorker.Put(w.db, messageStoreName, messageObj)
- if err != nil && !strings.Contains(err.Error(),
- "at least one key does not satisfy the uniqueness requirements") {
- // Only return non-unique constraint errors so that the case
- // below this one can be hit and handle duplicate entries properly.
- return 0, errors.Errorf("Unable to put Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- // NOTE: Sometimes the insert fails to return an error but hits a duplicate
- // insert, so this fallthrough returns the UUID entry in that case.
- if result.IsUndefined() {
- msgID := message.ID{}
- copy(msgID[:], newMessage.MessageID)
- msg, errLookup := w.msgIDLookup(msgID)
- if errLookup == nil && msg.ID != 0 {
- return msg.ID, nil
- }
- return 0, errors.Errorf("uuid lookup failure: %+v", err)
- }
- uuid := uint64(result.Int())
- jww.DEBUG.Printf("Successfully stored message %d", uuid)
+ return 0
+}
- return uuid, nil
+// GetMessageMessage is JSON marshalled and sent to the worker for
+// [wasmModel.GetMessage].
+type GetMessageMessage struct {
+ Message channels.ModelMessage `json:"message"`
+ Error string `json:"error"`
}
// GetMessage returns the message with the given [channel.MessageID].
func (w *wasmModel) GetMessage(
messageID message.ID) (channels.ModelMessage, error) {
- lookupResult, err := w.msgIDLookup(messageID)
- if err != nil {
- return channels.ModelMessage{}, err
- }
-
- var channelId *id.ID
- if lookupResult.ChannelID != nil {
- channelId, err = id.Unmarshal(lookupResult.ChannelID)
+ msgChan := make(chan GetMessageMessage)
+ w.wh.SendMessage(indexedDb.GetMessageTag, messageID.Marshal(), func(data []byte) {
+ var msg GetMessageMessage
+ err := json.Unmarshal(data, &msg)
if err != nil {
- return channels.ModelMessage{}, err
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to GetMessage: %+v", err)
}
- }
+ msgChan <- msg
+ })
- var parentMsgId message.ID
- if lookupResult.ParentMessageID != nil {
- parentMsgId, err = message.UnmarshalID(lookupResult.ParentMessageID)
- if err != nil {
- return channels.ModelMessage{}, err
+ select {
+ case msg := <-msgChan:
+ if msg.Error != "" {
+ return channels.ModelMessage{}, errors.New(msg.Error)
}
+ return msg.Message, nil
+ case <-time.After(indexedDb.ResponseTimeout):
+ return channels.ModelMessage{}, errors.Errorf("timed out after %s "+
+ "waiting for response from the worker about GetMessage",
+ indexedDb.ResponseTimeout)
}
-
- return channels.ModelMessage{
- UUID: lookupResult.ID,
- Nickname: lookupResult.Nickname,
- MessageID: messageID,
- ChannelID: channelId,
- ParentMessageID: parentMsgId,
- Timestamp: lookupResult.Timestamp,
- Lease: lookupResult.Lease,
- Status: channels.SentStatus(lookupResult.Status),
- Hidden: lookupResult.Hidden,
- Pinned: lookupResult.Pinned,
- Content: lookupResult.Text,
- Type: channels.MessageType(lookupResult.Type),
- Round: id.Round(lookupResult.Round),
- PubKey: lookupResult.Pubkey,
- CodesetVersion: lookupResult.CodesetVersion,
- }, nil
}
// DeleteMessage removes a message with the given messageID from storage.
func (w *wasmModel) DeleteMessage(messageID message.ID) error {
- msgId := js.ValueOf(base64.StdEncoding.EncodeToString(messageID.Bytes()))
- return indexedDbWorker.DeleteIndex(w.db, messageStoreName,
- messageStoreMessageIndex, pkeyName, msgId)
-}
-
-// msgIDLookup gets the UUID of the Message with the given messageID.
-func (w *wasmModel) msgIDLookup(messageID message.ID) (*Message, error) {
- msgIDStr := js.ValueOf(base64.StdEncoding.EncodeToString(messageID.Bytes()))
- resultObj, err := indexedDbWorker.GetIndex(w.db, messageStoreName,
- messageStoreMessageIndex, msgIDStr)
- if err != nil {
- return nil, err
- } else if resultObj.IsUndefined() {
- return nil, errors.Errorf("no message for %s found", msgIDStr)
- }
+ errChan := make(chan error)
+ w.wh.SendMessage(indexedDb.DeleteMessageTag, messageID.Marshal(), func(data []byte) {
+ if data != nil {
+ errChan <- errors.New(string(data))
+ } else {
+ errChan <- nil
+ }
+ })
- // Process result into string
- resultMsg := &Message{}
- err = json.Unmarshal([]byte(utils.JsToJson(resultObj)), resultMsg)
- if err != nil {
- return nil, err
+ select {
+ case err := <-errChan:
+ return err
+ case <-time.After(indexedDb.ResponseTimeout):
+ return errors.Errorf("timed out after %s waiting for response from "+
+ "the worker about DeleteMessage", indexedDb.ResponseTimeout)
}
- return resultMsg, nil
-
}
diff --git a/indexedDbWorker/channels/init.go b/indexedDbWorker/channels/init.go
index a1eba82bae80dd8ad035669bd86def850b8dfb88..a1bc72917e6dac1fb1ec6c7543c5c797f2e625eb 100644
--- a/indexedDbWorker/channels/init.go
+++ b/indexedDbWorker/channels/init.go
@@ -7,248 +7,168 @@
//go:build js && wasm
-package main
+package channels
import (
- "github.com/hack-pad/go-indexeddb/idb"
+ "encoding/json"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/channels"
cryptoChannel "gitlab.com/elixxir/crypto/channel"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
"gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/storage"
"gitlab.com/xx_network/primitives/id"
- "syscall/js"
"time"
)
-const (
- // databaseSuffix is the suffix to be appended to the name of
- // the database.
- databaseSuffix = "_speakeasy"
-
- // currentVersion is the current version of the IndexDb
- // runtime. Used for migration purposes.
- currentVersion uint = 1
-)
+// WorkerJavascriptFileURL is the URL of the script the worker will execute to
+// launch the worker WASM binary. It must obey the same-origin policy.
+const WorkerJavascriptFileURL = "/integrations/assets/dmIndexedDbWorker.js"
// MessageReceivedCallback is called any time a message is received or updated.
//
// update is true if the row is old and was edited.
type MessageReceivedCallback func(uuid uint64, channelID *id.ID, update bool)
+// NewWASMEventModelBuilder returns an EventModelBuilder which allows
+// the channel manager to define the path but the callback is the same
+// across the board.
+func NewWASMEventModelBuilder(encryption cryptoChannel.Cipher,
+ cb MessageReceivedCallback) channels.EventModelBuilder {
+ fn := func(path string) (channels.EventModel, error) {
+ return NewWASMEventModel(path, encryption, cb)
+ }
+ return fn
+}
+
+// NewWASMEventModelMessage is JSON marshalled and sent to the worker for
+// [NewWASMEventModel].
+type NewWASMEventModelMessage struct {
+ Path string `json:"path"`
+ EncryptionJSON string `json:"encryptionJSON"`
+}
+
// NewWASMEventModel returns a [channels.EventModel] backed by a wasmModel.
// The name should be a base64 encoding of the users public key.
func NewWASMEventModel(path string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
- channels.EventModel, error) {
- databaseName := path + databaseSuffix
- return newWASMModel(databaseName, encryption, cb, storeEncryptionStatus)
-}
+ cb MessageReceivedCallback) (channels.EventModel, error) {
-// storeEncryptionStatusFn matches storage.StoreIndexedDbEncryptionStatus so
-// that the data can be sent between the worker and main thread.
-type storeEncryptionStatusFn func(
- databaseName string, encryptionStatus bool) (bool, error)
-
-// newWASMModel creates the given [idb.Database] and returns a wasmModel.
-func newWASMModel(databaseName string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
- *wasmModel, error) {
- // Attempt to open database object
- ctx, cancel := indexedDbWorker.NewContext()
- defer cancel()
- openRequest, err := idb.Global().Open(ctx, databaseName, currentVersion,
- func(db *idb.Database, oldVersion, newVersion uint) error {
- if oldVersion == newVersion {
- jww.INFO.Printf("IndexDb version is current: v%d",
- newVersion)
- return nil
- }
-
- jww.INFO.Printf("IndexDb upgrade required: v%d -> v%d",
- oldVersion, newVersion)
-
- if oldVersion == 0 && newVersion >= 1 {
- err := v1Upgrade(db)
- if err != nil {
- return err
- }
- oldVersion = 1
- }
-
- // if oldVersion == 1 && newVersion >= 2 { v2Upgrade(), oldVersion = 2 }
- return nil
- })
+ // TODO: bring in URL and name from caller
+ wh, err := indexedDb.NewWorkerHandler(
+ WorkerJavascriptFileURL, "channelsIndexedDb")
if err != nil {
return nil, err
}
- // Wait for database open to finish
- db, err := openRequest.Await(ctx)
- if err != nil {
- return nil, err
- }
+ // Register handler to manage messages for the MessageReceivedCallback
+ wh.RegisterHandler(indexedDb.GetMessageTag, indexedDb.InitID, false,
+ messageReceivedCallbackHandler(cb))
+
+ // Register handler to manage checking encryption status from local storage
+ wh.RegisterHandler(indexedDb.EncryptionStatusTag, indexedDb.InitID, false,
+ checkDbEncryptionStatusHandler(wh))
- // Save the encryption status to storage
- encryptionStatus := encryption != nil
- loadedEncryptionStatus, err :=
- storeEncryptionStatus(databaseName, encryptionStatus)
+ encryptionJSON, err := json.Marshal(encryption)
if err != nil {
return nil, err
}
- // Verify encryption status does not change
- if encryptionStatus != loadedEncryptionStatus {
- return nil, errors.New(
- "Cannot load database with different encryption status.")
- } else if !encryptionStatus {
- jww.WARN.Printf("IndexedDb encryption disabled!")
+ message := NewWASMEventModelMessage{
+ Path: path,
+ EncryptionJSON: string(encryptionJSON),
}
- // Attempt to ensure the database has been properly initialized
- openRequest, err = idb.Global().Open(ctx, databaseName, currentVersion,
- func(db *idb.Database, oldVersion, newVersion uint) error {
- return nil
- })
+ payload, err := json.Marshal(message)
if err != nil {
return nil, err
}
- // Wait for database open to finish
- db, err = openRequest.Await(ctx)
- if err != nil {
- return nil, err
- }
- wrapper := &wasmModel{db: db, receivedMessageCB: cb, cipher: encryption}
- return wrapper, nil
-}
+ errChan := make(chan string)
+ wh.SendMessage(indexedDb.NewWASMEventModelTag, payload, func(data []byte) {
+ errChan <- string(data)
+ })
-// v1Upgrade performs the v0 -> v1 database upgrade.
-//
-// This can never be changed without permanently breaking backwards
-// compatibility.
-func v1Upgrade(db *idb.Database) error {
- storeOpts := idb.ObjectStoreOptions{
- KeyPath: js.ValueOf(pkeyName),
- AutoIncrement: true,
- }
- indexOpts := idb.IndexOptions{
- Unique: false,
- MultiEntry: false,
+ select {
+ case workerErr := <-errChan:
+ if workerErr != "" {
+ return nil, errors.New(workerErr)
+ }
+ case <-time.After(indexedDb.ResponseTimeout):
+ return nil, errors.Errorf("timed out after %s waiting for indexedDB "+
+ "database in worker to intialize", indexedDb.ResponseTimeout)
}
- // Build Message ObjectStore and Indexes
- messageStore, err := db.CreateObjectStore(messageStoreName, storeOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreMessageIndex,
- js.ValueOf(messageStoreMessage),
- idb.IndexOptions{
- Unique: true,
- MultiEntry: false,
- })
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreChannelIndex,
- js.ValueOf(messageStoreChannel), indexOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreParentIndex,
- js.ValueOf(messageStoreParent), indexOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreTimestampIndex,
- js.ValueOf(messageStoreTimestamp), indexOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStorePinnedIndex,
- js.ValueOf(messageStorePinned), indexOpts)
- if err != nil {
- return err
- }
+ return &wasmModel{wh}, nil
+}
- // Build Channel ObjectStore
- _, err = db.CreateObjectStore(channelsStoreName, storeOpts)
- if err != nil {
- return err
- }
+// MessageReceivedCallbackMessage is JSON marshalled and received from the
+// worker for the [MessageReceivedCallback] callback.
+type MessageReceivedCallbackMessage struct {
+ UUID uint64 `json:"uuid"`
+ ChannelID *id.ID `json:"channelID"`
+ Update bool `json:"update"`
+}
- // Get the database name and save it to storage
- if databaseName, err2 := db.Name(); err2 != nil {
- return err2
- } else if err = storeDatabaseName(databaseName); err != nil {
- return err
+// messageReceivedCallbackHandler returns a handler to manage messages for the
+// MessageReceivedCallback.
+func messageReceivedCallbackHandler(cb MessageReceivedCallback) func(data []byte) {
+ return func(data []byte) {
+ var msg MessageReceivedCallbackMessage
+ err := json.Unmarshal(data, &msg)
+ if err != nil {
+ jww.ERROR.Printf("Failed to JSON unmarshal "+
+ "MessageReceivedCallback message from worker: %+v", err)
+ return
+ }
+ cb(msg.UUID, msg.ChannelID, msg.Update)
}
+}
- return nil
+// EncryptionStatusMessage is JSON marshalled and received from the worker when
+// the database checks if it is encrypted.
+type EncryptionStatusMessage struct {
+ DatabaseName string `json:"databaseName"`
+ EncryptionStatus bool `json:"encryptionStatus"`
}
-// hackTestDb is a horrible function that exists as the result of an extremely
-// long discussion about why initializing the IndexedDb sometimes silently
-// fails. It ultimately tries to prevent an unrecoverable situation by actually
-// inserting some nonsense data and then checking to see if it persists.
-// If this function still exists in 2023, god help us all. Amen.
-func (w *wasmModel) hackTestDb() error {
- testMessage := &Message{
- ID: 0,
- Nickname: "test",
- MessageID: id.DummyUser.Marshal(),
- }
- msgId, helper := w.receiveHelper(testMessage, false)
- if helper != nil {
- return helper
- }
- result, err := indexedDbWorker.Get(w.db, messageStoreName, js.ValueOf(msgId))
- if err != nil {
- return err
- }
- if result.IsUndefined() {
- return errors.Errorf("Failed to test db, record not present")
- }
- return nil
+// EncryptionStatusReply is JSON marshalled and sent to the worker is response
+// to the [EncryptionStatusMessage].
+type EncryptionStatusReply struct {
+ EncryptionStatus bool `json:"encryptionStatus"`
+ Error string `json:"error"`
}
-// storeDatabaseName sends the database name to storage.StoreIndexedDb in the
-// main thread to be stored in localstorage and waits for the error to be
-// returned.
-//
-// The function specified below is a placeholder until set by
-// registerDatabaseNameStore. registerDatabaseNameStore must be called before
-// storeDatabaseName.
-var storeDatabaseName = func(databaseName string) error { return nil }
-
-// RegisterDatabaseNameStore sets storeDatabaseName to send the database to
-// storage.StoreIndexedDb in the main thread when called and registers a handler
-// to listen for the response.
-func RegisterDatabaseNameStore(m *manager) {
- storeDatabaseNameResponseChan := make(chan []byte)
- // Register handler
- m.mh.RegisterHandler(indexedDb.StoreDatabaseNameTag, func(data []byte) []byte {
- storeDatabaseNameResponseChan <- data
- return nil
- })
+// checkDbEncryptionStatusHandler returns a handler to manage checking
+// encryption status from local storage.
+func checkDbEncryptionStatusHandler(wh *indexedDb.WorkerHandler) func(data []byte) {
+ return func(data []byte) {
+ // Unmarshal received message
+ var msg EncryptionStatusMessage
+ err := json.Unmarshal(data, &msg)
+ if err != nil {
+ jww.ERROR.Printf("Failed to JSON unmarshal "+
+ "EncryptionStatusMessage message from worker: %+v", err)
+ return
+ }
- storeDatabaseName = func(databaseName string) error {
- m.mh.SendResponse(indexedDb.StoreDatabaseNameTag, indexedDb.InitID,
- []byte(databaseName))
-
- // Wait for response
- select {
- case response := <-storeDatabaseNameResponseChan:
- if len(response) > 0 {
- return errors.New(string(response))
- }
- case <-time.After(indexedDb.ResponseTimeout):
- return errors.Errorf("timed out after %s waiting for "+
- "response about storing the database name in local "+
- "storage in the main thread", indexedDb.ResponseTimeout)
+ // Pass message values to storage
+ loadedEncryptionStatus, err := storage.StoreIndexedDbEncryptionStatus(
+ msg.DatabaseName, msg.EncryptionStatus)
+ var reply EncryptionStatusReply
+ if err != nil {
+ reply.Error = err.Error()
+ } else {
+ reply.EncryptionStatus = loadedEncryptionStatus
}
- return nil
+
+ // Return response
+ statusData, err := json.Marshal(reply)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Failed to JSON marshal EncryptionStatusReply: %+v", err)
+ return
+ }
+
+ wh.SendMessage(indexedDb.EncryptionStatusTag, statusData, nil)
}
}
diff --git a/indexedDbWorker/dm/implementation.go b/indexedDbWorker/dm/implementation.go
index 9c8f5a5be24b76fa344c3100371b1c660327dade..0bbdcfd77ad0d53294c7d7ed27c5dde4c14cdd63 100644
--- a/indexedDbWorker/dm/implementation.go
+++ b/indexedDbWorker/dm/implementation.go
@@ -7,375 +7,240 @@
//go:build js && wasm
-package main
+package channelEventModel
import (
"crypto/ed25519"
"encoding/json"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
- "strings"
- "sync"
- "syscall/js"
"time"
- "github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/cmix/rounds"
"gitlab.com/elixxir/client/v4/dm"
- "gitlab.com/elixxir/xxdk-wasm/utils"
- "gitlab.com/xx_network/primitives/id"
-
- "github.com/hack-pad/go-indexeddb/idb"
- cryptoChannel "gitlab.com/elixxir/crypto/channel"
"gitlab.com/elixxir/crypto/message"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
)
// wasmModel implements dm.EventModel interface, which uses the channels system
// passed an object that adheres to in order to get events on the channel.
type wasmModel struct {
- db *idb.Database
- cipher cryptoChannel.Cipher
- receivedMessageCB MessageReceivedCallback
- updateMux sync.Mutex
-}
-
-// joinConversation is used for joining new conversations.
-func (w *wasmModel) joinConversation(nickname string,
- pubKey ed25519.PublicKey, dmToken uint32, codeset uint8) error {
- parentErr := errors.New("failed to joinConversation")
-
- // Build object
- newConvo := Conversation{
- Pubkey: pubKey,
- Nickname: nickname,
- Token: dmToken,
- CodesetVersion: codeset,
- Blocked: false,
- }
-
- // Convert to jsObject
- newConvoJson, err := json.Marshal(&newConvo)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to marshal Conversation: %+v", err)
- }
- convoObj, err := utils.JsonToJS(newConvoJson)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to marshal Conversation: %+v", err)
- }
-
- _, err = indexedDbWorker.Put(w.db, conversationStoreName, convoObj)
- if err != nil {
- return errors.WithMessagef(parentErr,
- "Unable to put Conversation: %+v", err)
- }
- return nil
+ wh *indexedDb.WorkerHandler
}
-// buildMessage is a private helper that converts typical dm.EventModel inputs
-// into a basic Message structure for insertion into storage.
-//
-// NOTE: ID is not set inside this function because we want to use the
-// autoincrement key by default. If you are trying to overwrite an existing
-// message, then you need to set it manually yourself.
-func buildMessage(messageID, parentID, text []byte, pubKey ed25519.PublicKey,
- timestamp time.Time, round id.Round, mType dm.MessageType,
- status dm.Status) *Message {
- return &Message{
- MessageID: messageID,
- ConversationPubKey: pubKey,
- ParentMessageID: parentID,
- Timestamp: timestamp,
- Status: uint8(status),
- Text: text,
- Type: uint16(mType),
- Round: uint64(round),
- }
+// TransferMessage is JSON marshalled and sent to the worker.
+type TransferMessage struct {
+ UUID uint64 `json:"uuid"`
+ MessageID message.ID `json:"messageID"`
+ ReactionTo message.ID `json:"reactionTo"`
+ Nickname string `json:"nickname"`
+ Text []byte `json:"text"`
+ PubKey ed25519.PublicKey `json:"pubKey"`
+ DmToken uint32 `json:"dmToken"`
+ Codeset uint8 `json:"codeset"`
+ Timestamp time.Time `json:"timestamp"`
+ Round rounds.Round `json:"round"`
+ MType dm.MessageType `json:"mType"`
+ Status dm.Status `json:"status"`
}
func (w *wasmModel) Receive(messageID message.ID, nickname string, text []byte,
pubKey ed25519.PublicKey, dmToken uint32, codeset uint8, timestamp time.Time,
round rounds.Round, mType dm.MessageType, status dm.Status) uint64 {
- parentErr := errors.New("failed to Receive")
-
- // If there is no extant Conversation, create one.
- _, err := indexedDbWorker.Get(
- w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
- if err != nil {
- if strings.Contains(err.Error(), indexedDbWorker.ErrDoesNotExist) {
- err = w.joinConversation(nickname, pubKey, dmToken, codeset)
- if err != nil {
- jww.ERROR.Printf("%+v", err)
- }
- } else {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get Conversation: %+v", err))
- }
+ msg := TransferMessage{
+ MessageID: messageID,
+ Nickname: nickname,
+ Text: text,
+ PubKey: pubKey,
+ DmToken: dmToken,
+ Codeset: codeset,
+ Timestamp: timestamp,
+ Round: round,
+ MType: mType,
+ Status: status,
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for TransferMessage: %+v", err)
return 0
- } else {
- jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- // Handle encryption, if it is present
- if w.cipher != nil {
- text, err = w.cipher.Encrypt(text)
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to Receive: %+v", err)
+ uuidChan <- 0
}
- }
+ uuidChan <- uuid
+ })
- msgToInsert := buildMessage(messageID.Bytes(), nil, text, pubKey, timestamp,
- round.ID, mType, status)
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about Receive", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, pubKey, false)
- return uuid
+ return 0
}
func (w *wasmModel) ReceiveText(messageID message.ID, nickname, text string,
pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- parentErr := errors.New("failed to ReceiveText")
-
- // If there is no extant Conversation, create one.
- _, err := indexedDbWorker.Get(
- w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
- if err != nil {
- if strings.Contains(err.Error(), indexedDbWorker.ErrDoesNotExist) {
- err = w.joinConversation(nickname, pubKey, dmToken, codeset)
- if err != nil {
- jww.ERROR.Printf("%+v", err)
- }
- } else {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get Conversation: %+v", err))
- }
+ msg := TransferMessage{
+ MessageID: messageID,
+ Nickname: nickname,
+ Text: []byte(text),
+ PubKey: pubKey,
+ DmToken: dmToken,
+ Codeset: codeset,
+ Timestamp: timestamp,
+ Round: round,
+ Status: status,
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for TransferMessage: %+v", err)
return 0
- } else {
- jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- // Handle encryption, if it is present
- textBytes := []byte(text)
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt(textBytes)
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveTextTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveText: %+v", err)
+ uuidChan <- 0
}
- }
+ uuidChan <- uuid
+ })
- msgToInsert := buildMessage(messageID.Bytes(), nil, textBytes,
- pubKey, timestamp, round.ID, dm.TextType, status)
-
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveText", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, pubKey, false)
- return uuid
+ return 0
}
func (w *wasmModel) ReceiveReply(messageID, reactionTo message.ID, nickname,
text string, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- parentErr := errors.New("failed to ReceiveReply")
-
- // If there is no extant Conversation, create one.
- _, err := indexedDbWorker.Get(
- w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
- if err != nil {
- if strings.Contains(err.Error(), indexedDbWorker.ErrDoesNotExist) {
- err = w.joinConversation(nickname, pubKey, dmToken, codeset)
- if err != nil {
- jww.ERROR.Printf("%+v", err)
- }
- } else {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get Conversation: %+v", err))
- }
+ msg := TransferMessage{
+ MessageID: messageID,
+ ReactionTo: reactionTo,
+ Nickname: nickname,
+ Text: []byte(text),
+ PubKey: pubKey,
+ DmToken: dmToken,
+ Codeset: codeset,
+ Timestamp: timestamp,
+ Round: round,
+ Status: status,
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for TransferMessage: %+v", err)
return 0
- } else {
- jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- // Handle encryption, if it is present
- textBytes := []byte(text)
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt(textBytes)
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveReplyTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveReply: %+v", err)
+ uuidChan <- 0
}
- }
-
- msgToInsert := buildMessage(messageID.Bytes(), reactionTo.Marshal(), textBytes,
- pubKey, timestamp, round.ID, dm.TextType, status)
+ uuidChan <- uuid
+ })
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveReply", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, pubKey, false)
- return uuid
+ return 0
}
-func (w *wasmModel) ReceiveReaction(messageID, _ message.ID, nickname,
+func (w *wasmModel) ReceiveReaction(messageID, reactionTo message.ID, nickname,
reaction string, pubKey ed25519.PublicKey, dmToken uint32, codeset uint8,
timestamp time.Time, round rounds.Round, status dm.Status) uint64 {
- parentErr := errors.New("failed to ReceiveText")
-
- // If there is no extant Conversation, create one.
- _, err := indexedDbWorker.Get(
- w.db, conversationStoreName, utils.CopyBytesToJS(pubKey))
- if err != nil {
- if strings.Contains(err.Error(), indexedDbWorker.ErrDoesNotExist) {
- err = w.joinConversation(nickname, pubKey, dmToken, codeset)
- if err != nil {
- jww.ERROR.Printf("%+v", err)
- }
- } else {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get Conversation: %+v", err))
- }
+ msg := TransferMessage{
+ MessageID: messageID,
+ ReactionTo: reactionTo,
+ Nickname: nickname,
+ Text: []byte(reaction),
+ PubKey: pubKey,
+ DmToken: dmToken,
+ Codeset: codeset,
+ Timestamp: timestamp,
+ Round: round,
+ Status: status,
+ }
+
+ data, err := json.Marshal(msg)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for TransferMessage: %+v", err)
return 0
- } else {
- jww.DEBUG.Printf("Conversation with %s already joined", nickname)
}
- // Handle encryption, if it is present
- textBytes := []byte(reaction)
- if w.cipher != nil {
- textBytes, err = w.cipher.Encrypt(textBytes)
+ uuidChan := make(chan uint64)
+ w.wh.SendMessage(indexedDb.ReceiveReactionTag, data, func(data []byte) {
+ var uuid uint64
+ err = json.Unmarshal(data, &uuid)
if err != nil {
- jww.ERROR.Printf("Failed to encrypt Message: %+v", err)
- return 0
+ jww.ERROR.Printf(
+ "Could not JSON unmarshal response to ReceiveReaction: %+v", err)
+ uuidChan <- 0
}
- }
-
- msgToInsert := buildMessage(messageID.Bytes(), nil, textBytes,
- pubKey, timestamp, round.ID, dm.ReactionType, status)
+ uuidChan <- uuid
+ })
- uuid, err := w.receiveHelper(msgToInsert, false)
- if err != nil {
- jww.ERROR.Printf("Failed to receive Message: %+v", err)
+ select {
+ case uuid := <-uuidChan:
+ return uuid
+ case <-time.After(indexedDb.ResponseTimeout):
+ jww.ERROR.Printf("Timed out after %s waiting for response from the "+
+ "worker about ReceiveReaction", indexedDb.ResponseTimeout)
}
- go w.receivedMessageCB(uuid, pubKey, false)
- return uuid
+ return 0
}
func (w *wasmModel) UpdateSentStatus(uuid uint64, messageID message.ID,
timestamp time.Time, round rounds.Round, status dm.Status) {
- parentErr := errors.New("failed to UpdateSentStatus")
-
- // FIXME: this is a bit of race condition without the mux.
- // This should be done via the transactions (i.e., make a
- // special version of receiveHelper)
- w.updateMux.Lock()
- defer w.updateMux.Unlock()
-
- // Convert messageID to the key generated by json.Marshal
- key := js.ValueOf(uuid)
-
- // Use the key to get the existing Message
- currentMsg, err := indexedDbWorker.Get(w.db, messageStoreName, key)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Unable to get message: %+v", err))
- return
- }
-
- // Extract the existing Message and update the Status
- newMessage := &Message{}
- err = json.Unmarshal([]byte(utils.JsToJson(currentMsg)), newMessage)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.WithMessagef(parentErr,
- "Could not JSON unmarshal message: %+v", err))
- return
- }
-
- newMessage.Status = uint8(status)
- if !messageID.Equals(message.ID{}) {
- newMessage.MessageID = messageID.Bytes()
- }
-
- if round.ID != 0 {
- newMessage.Round = uint64(round.ID)
- }
-
- if !timestamp.Equal(time.Time{}) {
- newMessage.Timestamp = timestamp
- }
-
- // Store the updated Message
- _, err = w.receiveHelper(newMessage, true)
- if err != nil {
- jww.ERROR.Printf("%+v", errors.Wrap(parentErr, err.Error()))
- }
- go w.receivedMessageCB(uuid, newMessage.ConversationPubKey, true)
-}
-
-// receiveHelper is a private helper for receiving any sort of message.
-func (w *wasmModel) receiveHelper(
- newMessage *Message, isUpdate bool) (uint64, error) {
- // Convert to jsObject
- newMessageJson, err := json.Marshal(newMessage)
- if err != nil {
- return 0, errors.Errorf("Unable to marshal Message: %+v", err)
- }
- messageObj, err := utils.JsonToJS(newMessageJson)
- if err != nil {
- return 0, errors.Errorf("Unable to marshal Message: %+v", err)
- }
-
- // Unset the primaryKey for inserts so that it can be auto-populated and
- // incremented
- if !isUpdate {
- messageObj.Delete("id")
+ msg := TransferMessage{
+ UUID: uuid,
+ MessageID: messageID,
+ Timestamp: timestamp,
+ Round: round,
+ Status: status,
}
- // Store message to database
- result, err := indexedDbWorker.Put(w.db, messageStoreName, messageObj)
+ data, err := json.Marshal(msg)
if err != nil {
- return 0, errors.Errorf("Unable to put Message: %+v", err)
+ jww.ERROR.Printf(
+ "Could not JSON marshal payload for TransferMessage: %+v", err)
}
- // NOTE: Sometimes the insert fails to return an error but hits a duplicate
- // insert, so this fallthrough returns the UUID entry in that case.
- if result.IsUndefined() {
- msgID := message.ID{}
- copy(msgID[:], newMessage.MessageID)
- uuid, errLookup := w.msgIDLookup(msgID)
- if uuid != 0 && errLookup == nil {
- return uuid, nil
- }
- return 0, errors.Errorf("uuid lookup failure: %+v", err)
- }
- uuid := uint64(result.Int())
- jww.DEBUG.Printf("Successfully stored message %d", uuid)
-
- return uuid, nil
-}
-
-// msgIDLookup gets the UUID of the Message with the given messageID.
-func (w *wasmModel) msgIDLookup(messageID message.ID) (uint64, error) {
- resultObj, err := indexedDbWorker.GetIndex(w.db, messageStoreName,
- messageStoreMessageIndex, utils.CopyBytesToJS(messageID.Marshal()))
- if err != nil {
- return 0, err
- }
-
- uuid := uint64(0)
- if !resultObj.IsUndefined() {
- uuid = uint64(resultObj.Get("id").Int())
- }
- return uuid, nil
+ w.wh.SendMessage(indexedDb.UpdateSentStatusTag, data, nil)
}
diff --git a/indexedDbWorker/dm/init.go b/indexedDbWorker/dm/init.go
index 184f81311a20f4c0e5f0cfe7a802cd8df4d1bed3..269d7d135ca87331516c364ef2983240d076e552 100644
--- a/indexedDbWorker/dm/init.go
+++ b/indexedDbWorker/dm/init.go
@@ -7,31 +7,24 @@
//go:build js && wasm
-package main
+package channelEventModel
import (
"crypto/ed25519"
- "gitlab.com/elixxir/xxdk-wasm/indexedDb"
- "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
- "syscall/js"
+ "encoding/json"
"time"
- "github.com/hack-pad/go-indexeddb/idb"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/v4/dm"
cryptoChannel "gitlab.com/elixxir/crypto/channel"
+ "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker"
+ "gitlab.com/elixxir/xxdk-wasm/storage"
)
-const (
- // databaseSuffix is the suffix to be appended to the name of
- // the database.
- databaseSuffix = "_speakeasy_dm"
-
- // currentVersion is the current version of the IndexDb
- // runtime. Used for migration purposes.
- currentVersion uint = 1
-)
+// WorkerJavascriptFileURL is the URL of the script the worker will execute to
+// launch the worker WASM binary. It must obey the same-origin policy.
+const WorkerJavascriptFileURL = "/integrations/assets/indexedDbWorker.js"
// MessageReceivedCallback is called any time a message is received or updated.
//
@@ -39,192 +32,134 @@ const (
type MessageReceivedCallback func(
uuid uint64, pubKey ed25519.PublicKey, update bool)
+// NewWASMEventModelMessage is JSON marshalled and sent to the worker for
+// [NewWASMEventModel].
+type NewWASMEventModelMessage struct {
+ Path string `json:"path"`
+ EncryptionJSON string `json:"encryptionJSON"`
+}
+
// NewWASMEventModel returns a [channels.EventModel] backed by a wasmModel.
// The name should be a base64 encoding of the users public key.
func NewWASMEventModel(path string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
- dm.EventModel, error) {
- databaseName := path + databaseSuffix
- return newWASMModel(databaseName, encryption, cb, storeEncryptionStatus)
-}
+ cb MessageReceivedCallback) (dm.EventModel, error) {
-// storeEncryptionStatusFn matches storage.StoreIndexedDbEncryptionStatus so
-// that the data can be sent between the worker and main thread.
-type storeEncryptionStatusFn func(
- databaseName string, encryptionStatus bool) (bool, error)
-
-// newWASMModel creates the given [idb.Database] and returns a wasmModel.
-func newWASMModel(databaseName string, encryption cryptoChannel.Cipher,
- cb MessageReceivedCallback, storeEncryptionStatus storeEncryptionStatusFn) (
- *wasmModel, error) {
- // Attempt to open database object
- ctx, cancel := indexedDbWorker.NewContext()
- defer cancel()
- openRequest, err := idb.Global().Open(ctx, databaseName, currentVersion,
- func(db *idb.Database, oldVersion, newVersion uint) error {
- if oldVersion == newVersion {
- jww.INFO.Printf("IndexDb version is current: v%d",
- newVersion)
- return nil
- }
-
- jww.INFO.Printf("IndexDb upgrade required: v%d -> v%d",
- oldVersion, newVersion)
-
- if oldVersion == 0 && newVersion >= 1 {
- err := v1Upgrade(db)
- if err != nil {
- return err
- }
- oldVersion = 1
- }
-
- // if oldVersion == 1 && newVersion >= 2 { v2Upgrade(), oldVersion = 2 }
- return nil
- })
+ // TODO: bring in URL and name from caller
+ wh, err := indexedDb.NewWorkerHandler(
+ WorkerJavascriptFileURL, "dmIndexedDb")
if err != nil {
return nil, err
}
- // Wait for database open to finish
- db, err := openRequest.Await(ctx)
- if err != nil {
- return nil, err
- }
+ // Register handler to manage messages for the MessageReceivedCallback
+ wh.RegisterHandler(indexedDb.GetMessageTag, indexedDb.InitID, false,
+ messageReceivedCallbackHandler(cb))
+
+ // Register handler to manage checking encryption status from local storage
+ wh.RegisterHandler(indexedDb.EncryptionStatusTag, indexedDb.InitID, false,
+ checkDbEncryptionStatusHandler(wh))
- // Save the encryption status to storage
- encryptionStatus := encryption != nil
- loadedEncryptionStatus, err :=
- storeEncryptionStatus(databaseName, encryptionStatus)
+ encryptionJSON, err := json.Marshal(encryption)
if err != nil {
return nil, err
}
- // Verify encryption status does not change
- if encryptionStatus != loadedEncryptionStatus {
- return nil, errors.New(
- "Cannot load database with different encryption status.")
- } else if !encryptionStatus {
- jww.WARN.Printf("IndexedDb encryption disabled!")
+ message := NewWASMEventModelMessage{
+ Path: path,
+ EncryptionJSON: string(encryptionJSON),
}
- // Attempt to ensure the database has been properly initialized
- openRequest, err = idb.Global().Open(ctx, databaseName, currentVersion,
- func(db *idb.Database, oldVersion, newVersion uint) error {
- return nil
- })
+ payload, err := json.Marshal(message)
if err != nil {
return nil, err
}
- // Wait for database open to finish
- db, err = openRequest.Await(ctx)
- if err != nil {
- return nil, err
- }
- wrapper := &wasmModel{db: db, receivedMessageCB: cb, cipher: encryption}
- return wrapper, nil
-}
+ errChan := make(chan string)
+ wh.SendMessage(indexedDb.NewWASMEventModelTag, payload, func(data []byte) {
+ errChan <- string(data)
+ })
-// v1Upgrade performs the v0 -> v1 database upgrade.
-//
-// This can never be changed without permanently breaking backwards
-// compatibility.
-func v1Upgrade(db *idb.Database) error {
- indexOpts := idb.IndexOptions{
- Unique: false,
- MultiEntry: false,
+ select {
+ case workerErr := <-errChan:
+ if workerErr != "" {
+ return nil, errors.New(workerErr)
+ }
+ case <-time.After(indexedDb.ResponseTimeout):
+ return nil, errors.Errorf("timed out after %s waiting for indexedDB "+
+ "database in worker to intialize", indexedDb.ResponseTimeout)
}
- // Build Message ObjectStore and Indexes
- messageStoreOpts := idb.ObjectStoreOptions{
- KeyPath: js.ValueOf(msgPkeyName),
- AutoIncrement: true,
- }
- messageStore, err := db.CreateObjectStore(messageStoreName, messageStoreOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreMessageIndex,
- js.ValueOf(messageStoreMessage),
- idb.IndexOptions{
- Unique: true,
- MultiEntry: false,
- })
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreConversationIndex,
- js.ValueOf(messageStoreConversation), indexOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreParentIndex,
- js.ValueOf(messageStoreParent), indexOpts)
- if err != nil {
- return err
- }
- _, err = messageStore.CreateIndex(messageStoreTimestampIndex,
- js.ValueOf(messageStoreTimestamp), indexOpts)
- if err != nil {
- return err
- }
+ return &wasmModel{wh}, nil
+}
- // Build Channel ObjectStore
- conversationStoreOpts := idb.ObjectStoreOptions{
- KeyPath: js.ValueOf(convoPkeyName),
- AutoIncrement: false,
- }
- _, err = db.CreateObjectStore(conversationStoreName, conversationStoreOpts)
- if err != nil {
- return err
- }
+// MessageReceivedCallbackMessage is JSON marshalled and received from the
+// worker for the [MessageReceivedCallback] callback.
+type MessageReceivedCallbackMessage struct {
+ UUID uint64 `json:"uuid"`
+ PubKey ed25519.PublicKey `json:"pubKey"`
+ Update bool `json:"update"`
+}
- // Get the database name and save it to storage
- if databaseName, err2 := db.Name(); err2 != nil {
- return err2
- } else if err = storeDatabaseName(databaseName); err != nil {
- return err
+// messageReceivedCallbackHandler returns a handler to manage messages for the
+// MessageReceivedCallback.
+func messageReceivedCallbackHandler(cb MessageReceivedCallback) func(data []byte) {
+ return func(data []byte) {
+ var msg MessageReceivedCallbackMessage
+ err := json.Unmarshal(data, &msg)
+ if err != nil {
+ jww.ERROR.Printf("Failed to JSON unmarshal "+
+ "MessageReceivedCallback message from worker: %+v", err)
+ return
+ }
+ cb(msg.UUID, msg.PubKey, msg.Update)
}
+}
- return nil
+// EncryptionStatusMessage is JSON marshalled and received from the worker when
+// the database checks if it is encrypted.
+type EncryptionStatusMessage struct {
+ DatabaseName string `json:"databaseName"`
+ EncryptionStatus bool `json:"encryptionStatus"`
}
-// storeDatabaseName sends the database name to storage.StoreIndexedDb in the
-// main thread to be stored in localstorage and waits for the error to be
-// returned.
-//
-// The function specified below is a placeholder until set by
-// registerDatabaseNameStore. registerDatabaseNameStore must be called before
-// storeDatabaseName.
-var storeDatabaseName = func(databaseName string) error { return nil }
-
-// RegisterDatabaseNameStore sets storeDatabaseName to send the database to
-// storage.StoreIndexedDb in the main thread when called and registers a handler
-// to listen for the response.
-func RegisterDatabaseNameStore(m *manager) {
- storeDatabaseNameResponseChan := make(chan []byte)
- // Register handler
- m.mh.RegisterHandler(indexedDb.StoreDatabaseNameTag, func(data []byte) []byte {
- storeDatabaseNameResponseChan <- data
- return nil
- })
+// EncryptionStatusReply is JSON marshalled and sent to the worker is response
+// to the [EncryptionStatusMessage].
+type EncryptionStatusReply struct {
+ EncryptionStatus bool `json:"encryptionStatus"`
+ Error string `json:"error"`
+}
- storeDatabaseName = func(databaseName string) error {
- m.mh.SendResponse(indexedDb.StoreDatabaseNameTag, indexedDb.InitID,
- []byte(databaseName))
-
- // Wait for response
- select {
- case response := <-storeDatabaseNameResponseChan:
- if len(response) > 0 {
- return errors.New(string(response))
- }
- case <-time.After(indexedDb.ResponseTimeout):
- return errors.Errorf("timed out after %s waiting for "+
- "response about storing the database name in local "+
- "storage in the main thread", indexedDb.ResponseTimeout)
+// checkDbEncryptionStatusHandler returns a handler to manage checking
+// encryption status from local storage.
+func checkDbEncryptionStatusHandler(wh *indexedDb.WorkerHandler) func(data []byte) {
+ return func(data []byte) {
+ // Unmarshal received message
+ var msg EncryptionStatusMessage
+ err := json.Unmarshal(data, &msg)
+ if err != nil {
+ jww.ERROR.Printf("Failed to JSON unmarshal "+
+ "EncryptionStatusMessage message from worker: %+v", err)
+ return
}
- return nil
+
+ // Pass message values to storage
+ loadedEncryptionStatus, err := storage.StoreIndexedDbEncryptionStatus(
+ msg.DatabaseName, msg.EncryptionStatus)
+ var reply EncryptionStatusReply
+ if err != nil {
+ reply.Error = err.Error()
+ } else {
+ reply.EncryptionStatus = loadedEncryptionStatus
+ }
+
+ // Return response
+ statusData, err := json.Marshal(reply)
+ if err != nil {
+ jww.ERROR.Printf(
+ "Failed to JSON marshal EncryptionStatusReply: %+v", err)
+ return
+ }
+
+ wh.SendMessage(indexedDb.EncryptionStatusTag, statusData, nil)
}
}
diff --git a/indexedDb/tag.go b/indexedDbWorker/tag.go
similarity index 100%
rename from indexedDb/tag.go
rename to indexedDbWorker/tag.go
diff --git a/indexedDb/worker.go b/indexedDbWorker/worker.go
similarity index 100%
rename from indexedDb/worker.go
rename to indexedDbWorker/worker.go
diff --git a/wasm/channels.go b/wasm/channels.go
index d32737321c0efbab9ff732917b185a7b2cfbea99..5fa1edced65da86700f88414a6b4d04e0c91aa7b 100644
--- a/wasm/channels.go
+++ b/wasm/channels.go
@@ -14,7 +14,7 @@ import (
"encoding/json"
"errors"
"gitlab.com/elixxir/client/v4/channels"
- channelsDb "gitlab.com/elixxir/xxdk-wasm/indexedDb/channels"
+ channelsDb "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker/channels"
"gitlab.com/xx_network/primitives/id"
"sync"
"syscall/js"
@@ -1832,7 +1832,7 @@ func (em *eventModel) GetMessage(messageID []byte) ([]byte, error) {
// the database.
//
// Parameters:
-// - messageID - The bytes of the [channel.MessageID] of the message.
+// - messageID - The bytes of the [channel.MessageID] of the message.
func (em *eventModel) DeleteMessage(messageID []byte) error {
err := em.deleteMessage(utils.CopyBytesToJS(messageID))
if !err.IsUndefined() {
diff --git a/wasm/dm.go b/wasm/dm.go
index 678ca74f529f1ccce1a0a8cac63242b5073ed835..613835eb6490fc77e1189bc213c57b39c71f9c88 100644
--- a/wasm/dm.go
+++ b/wasm/dm.go
@@ -13,7 +13,7 @@ import (
"crypto/ed25519"
"syscall/js"
- indexDB "gitlab.com/elixxir/xxdk-wasm/indexedDb/dm"
+ indexDB "gitlab.com/elixxir/xxdk-wasm/indexedDbWorker/dm"
"encoding/base64"
@@ -129,7 +129,7 @@ func NewDMClient(_ js.Value, args []js.Value) any {
}
// NewDMClientWithIndexedDb creates a new [DMClient] from a new
-// private identity ([channel.PrivateIdentity]) and using indexedDb as a backend
+// private identity ([channel.PrivateIdentity]) and using indexedDbWorker as a backend
// to manage the event model.
//
// This is for creating a manager for an identity for the first time. For
@@ -137,7 +137,7 @@ func NewDMClient(_ js.Value, args []js.Value) any {
// reload this channel manager, use [LoadDMClientWithIndexedDb], passing
// in the storage tag retrieved by [DMClient.GetStorageTag].
//
-// This function initialises an indexedDb database.
+// This function initialises an indexedDbWorker database.
//
// Parameters:
// - args[0] - ID of [Cmix] object in tracker (int). This can be retrieved
@@ -145,7 +145,7 @@ func NewDMClient(_ js.Value, args []js.Value) any {
// - args[1] - Bytes of a private identity ([channel.PrivateIdentity]) that is
// generated by [GenerateChannelIdentity] (Uint8Array).
// - args[2] - Function that takes in the same parameters as
-// [indexedDb.MessageReceivedCallback]. On the Javascript side, the UUID is
+// [indexedDbWorker.MessageReceivedCallback]. On the Javascript side, the UUID is
// returned as an int and the channelID as a Uint8Array. The row in the
// database that was updated can be found using the UUID. The channel ID is
// provided so that the recipient can filter if they want to the processes
@@ -157,7 +157,7 @@ func NewDMClient(_ js.Value, args []js.Value) any {
//
// Returns a promise:
// - Resolves to a Javascript representation of the [DMClient] object.
-// - Rejected with an error if loading indexedDb or the manager fails.
+// - Rejected with an error if loading indexedDbWorker or the manager fails.
// - Throws a TypeError if the cipher ID does not correspond to a cipher.
func NewDMClientWithIndexedDb(_ js.Value, args []js.Value) any {
cmixID := args[0].Int()
@@ -175,7 +175,7 @@ func NewDMClientWithIndexedDb(_ js.Value, args []js.Value) any {
}
// NewDMClientWithIndexedDbUnsafe creates a new [DMClient] from a
-// new private identity ([channel.PrivateIdentity]) and using indexedDb as a
+// new private identity ([channel.PrivateIdentity]) and using indexedDbWorker as a
// backend to manage the event model. However, the data is written in plain text
// and not encrypted. It is recommended that you do not use this in production.
//
@@ -184,7 +184,7 @@ func NewDMClientWithIndexedDb(_ js.Value, args []js.Value) any {
// reload this channel manager, use [LoadDMClientWithIndexedDbUnsafe],
// passing in the storage tag retrieved by [DMClient.GetStorageTag].
//
-// This function initialises an indexedDb database.
+// This function initialises an indexedDbWorker database.
//
// Parameters:
// - args[0] - ID of [Cmix] object in tracker (int). This can be retrieved
@@ -192,7 +192,7 @@ func NewDMClientWithIndexedDb(_ js.Value, args []js.Value) any {
// - args[1] - Bytes of a private identity ([channel.PrivateIdentity]) that is
// generated by [GenerateChannelIdentity] (Uint8Array).
// - args[2] - Function that takes in the same parameters as
-// [indexedDb.MessageReceivedCallback]. On the Javascript side, the UUID is
+// [indexedDbWorker.MessageReceivedCallback]. On the Javascript side, the UUID is
// returned as an int and the channelID as a Uint8Array. The row in the
// database that was updated can be found using the UUID. The channel ID is
// provided so that the recipient can filter if they want to the processes
@@ -201,7 +201,7 @@ func NewDMClientWithIndexedDb(_ js.Value, args []js.Value) any {
//
// Returns a promise:
// - Resolves to a Javascript representation of the [DMClient] object.
-// - Rejected with an error if loading indexedDb or the manager fails.
+// - Rejected with an error if loading indexedDbWorker or the manager fails.
func NewDMClientWithIndexedDbUnsafe(_ js.Value, args []js.Value) any {
cmixID := args[0].Int()
privateIdentity := utils.CopyBytesToGo(args[1])