Allow Peers to Store History

.gitignore

@@ -9,5 +9,6 @@ server/app/messages
 .reviewboardrc
 /vendor/
 /testing/tor/
+/storage/*/testing/
 /storage/testing/
 /testing/storage/

event/bridge/pipeBridge_test.go

@@ -93,7 +93,7 @@ func stableService(t *testing.T, in, out string, done chan bool) {
 		return
 	}
 	if message1.Message.EventType != event.NewPeer {
-		t.Errorf("Wrong message recieved, expected NewPeer\n")
+		t.Errorf("Wrong message received, expected NewPeer\n")
 		done <- true
 		return
 	}
@@ -107,7 +107,7 @@ func stableService(t *testing.T, in, out string, done chan bool) {
 		return
 	}
 	if message2.Message.EventType != event.DeleteContact {
-		t.Errorf("Wrong message recieved, expected DeleteContact, got %v\n", message2)
+		t.Errorf("Wrong message received, expected DeleteContact, got %v\n", message2)
 		done <- true
 		return
 	}

event/common.go

@@ -266,3 +266,18 @@ const (
 	ContextGetVal = "im.cwtch.getVal"
 	ContextRetVal = "im.cwtch.retVal"
 )
+
+// Define Default Attribute Keys
+const (
+	SaveHistoryKey = "SavePeerHistory"
+)
+
+// Define Default Attribute Values
+const (
+	// Save History has 3 distinct states. By default we don't save history (DontSaveHistoryDefault), if the peer confirms this
+	// we change to DontSaveHistoryConfirmed, if they confirm they want to save then this becomes SaveHistoryConfirmed
+	// We use this distinction between default and confirmed to drive UI
+	DontSaveHistoryDefault   = "DefaultDontSaveHistory"
+	SaveHistoryConfirmed     = "SaveHistory"
+	DontSaveHistoryConfirmed = "DontSaveHistory"
+)

model/attr/scope.go

@@ -11,7 +11,7 @@ A local peer "Alice" has a PublicScope that is queryable by getVal requests.
 By default, for now, other scopes are private, of which we here define SettingsScope
 
 Alice's peer structs of remote peers such as "Bob" keep the queried
-PublicScope values in the PeerScope, which can be overriden by the same named
+PublicScope values in the PeerScope, which can be overridden by the same named
 values stored in the LocalScope.
 
 */
@@ -27,32 +27,32 @@ const (
 	SettingsScope = "settings"
 )
 
-// Seperator for scope and the rest of path
+// Separator for scope and the rest of path
-const Seperator = "."
+const Separator = "."
 
 // GetPublicScope takes a path and attaches the pubic scope to it
 func GetPublicScope(path string) string {
-	return PublicScope + Seperator + path
+	return PublicScope + Separator + path
 }
 
 // GetSettingsScope takes a path and attaches the settings scope to it
 func GetSettingsScope(path string) string {
-	return SettingsScope + Seperator + path
+	return SettingsScope + Separator + path
 }
 
 // GetLocalScope takes a path and attaches the local scope to it
 func GetLocalScope(path string) string {
-	return LocalScope + Seperator + path
+	return LocalScope + Separator + path
 }
 
 // GetPeerScope takes a path and attaches the peer scope to it
 func GetPeerScope(path string) string {
-	return PeerScope + Seperator + path
+	return PeerScope + Separator + path
 }
 
 // GetScopePath take a full path and returns the scope and the scope-less path
 func GetScopePath(fullPath string) (string, string) {
-	parts := strings.SplitN(fullPath, Seperator, 1)
+	parts := strings.SplitN(fullPath, Separator, 1)
 	if len(parts) != 2 {
 		return "", ""
 	}

model/profile.go

@@ -22,7 +22,7 @@ import (
 type Authorization string
 
 const (
-	// AuthUnknown is an inital state for a new unseen peer
+	// AuthUnknown is an initial state for a new unseen peer
 	AuthUnknown Authorization = "unknown"
 	// AuthApproved means the client has approved the peer, it can send messages to us, perform GetVals, etc
 	AuthApproved Authorization = "approved"
@@ -469,6 +469,10 @@ func (p *Profile) GetCopy(timeline bool) *Profile {
 		for groupID := range newp.Groups {
 			newp.Groups[groupID].Timeline = *p.Groups[groupID].Timeline.GetCopy()
 		}
+
+		for peerID := range newp.Contacts {
+			newp.Contacts[peerID].Timeline = *p.Contacts[peerID].Timeline.GetCopy()
+		}
 	}
 
 	return newp

storage/v0/stream_store.go

@@ -46,7 +46,7 @@ func NewStreamStore(directory string, filenameBase string, password string) (sto
 	return ss
 }
 
-// Read returns all messages from the backing file (not the buffer, which is jsut for writing to the current file)
+// Read returns all messages from the backing file (not the buffer, for writing to the current file)
 func (ss *streamStore) Read() (messages []model.Message) {
 	ss.lock.Lock()
 	defer ss.lock.Unlock()

storage/v1/profile_store.go

@@ -258,6 +258,10 @@ func (ps *ProfileStoreV1) load() error {
 					contact.Authorization = model.AuthApproved
 				}
 			}
+
+			ss := NewStreamStore(ps.directory, contact.LocalID, ps.key)
+			cp.Contacts[contact.Onion].Timeline.SetMessages(ss.Read())
+			ps.streamStores[contact.Onion] = ss
 		}
 
 		for gid, group := range cp.Groups {
@@ -266,6 +270,7 @@ func (ps *ProfileStoreV1) load() error {
 			cp.Groups[gid].Timeline.SetMessages(ss.Read())
 			ps.streamStores[group.GroupID] = ss
 		}
+
 	}
 
 	return err
@@ -292,11 +297,6 @@ func (ps *ProfileStoreV1) eventHandler() {
 			var pp *model.PublicProfile
 			json.Unmarshal([]byte(ev.Data[event.Data]), &pp)
 			ps.profile.AddContact(ev.Data[event.RemotePeer], pp)
-			// TODO: configure - allow peers to be configured to turn on limited storage
-			/*ss := NewStreamStore(ps.directory, pp.LocalID, ps.password)
-			pp.Timeline.SetMessages(ss.Read())
-			ps.streamStores[pp.Onion] = ss
-			ps.save()*/
 		case event.GroupCreated:
 			var group *model.Group
 			json.Unmarshal([]byte(ev.Data[event.Data]), &group)
@@ -343,6 +343,41 @@ func (ps *ProfileStoreV1) eventHandler() {
 			} else {
 				log.Errorf("error storing new group invite: %v (%v)", err, ev)
 			}
+		case event.NewMessageFromPeer:
+			contact, exists := ps.profile.GetContact(ev.Data[event.RemotePeer])
+			if exists {
+				val, ok := contact.GetAttribute(event.SaveHistoryKey)
+				if !ok {
+					contact.SetAttribute(event.SaveHistoryKey, event.DontSaveHistoryDefault)
+				} else {
+					switch val {
+					case event.SaveHistoryConfirmed:
+						{
+							peerID := ev.Data[event.RemotePeer]
+							received, _ := time.Parse(time.RFC3339Nano, ev.Data[event.TimestampReceived])
+							message := model.Message{Received: received, Timestamp: received, Message: ev.Data[event.Data], PeerID: peerID, Signature: []byte{}, PreviousMessageSig: []byte{}}
+							ss, exists := ps.streamStores[peerID]
+							if exists {
+								ss.Write(message)
+							} else {
+								log.Errorf("error storing new peer message: %v stream store does not exist", ev)
+								ss := NewStreamStore(ps.directory, contact.LocalID, ps.key)
+								ps.streamStores[peerID] = ss
+								ss.Write(message)
+							}
+						}
+					case event.DontSaveHistoryDefault:
+						{
+						}
+					case event.DontSaveHistoryConfirmed:
+						{
+						}
+					}
+				}
+				ps.save()
+			} else {
+				log.Errorf("error setting attribute on peer %v peer does not exist", ev)
+			}
 		case event.NewMessageFromGroup:
 			groupid := ev.Data[event.GroupID]
 			received, _ := time.Parse(time.RFC3339Nano, ev.Data[event.TimestampReceived])
@@ -368,6 +403,11 @@ func (ps *ProfileStoreV1) eventHandler() {
 			onion := ev.Data[event.RemotePeer]
 			ps.profile.DeleteContact(onion)
 			ps.save()
+			ss, exists := ps.streamStores[onion]
+			if exists {
+				ss.Delete()
+				delete(ps.streamStores, onion)
+			}
 		case event.DeleteGroup:
 			groupID := ev.Data[event.GroupID]
 			ps.profile.DeleteGroup(groupID)

storage/v1/stream_store.go

@@ -108,7 +108,7 @@ func (ss *streamStore) Delete() {
 	}
 }
 
-// Read returns all messages from the backing file (not the buffer, which is jsut for writing to the current file)
+// Read returns all messages from the backing file (not the buffer, for writing to the current file)
 func (ss *streamStore) Read() (messages []model.Message) {
 	ss.lock.Lock()
 	defer ss.lock.Unlock()

testing/cwtch_peer_server_integration_test.go

@@ -63,7 +63,7 @@ func serverCheck(t *testing.T, serverAddr string) bool {
 
 func waitForPeerGroupConnection(t *testing.T, peer peer.CwtchPeer, groupID string) {
 	for {
-		fmt.Printf("%v checking group conection...\n", peer.GetName())
+		fmt.Printf("%v checking group connection...\n", peer.GetName())
 		state, ok := peer.GetGroupState(groupID)
 		if ok {
 			fmt.Printf("Waiting for Peer %v to join group %v - state: %v\n", peer.GetName(), groupID, state)