cwtch/app/app.go

package app

import (
	"cwtch.im/cwtch/event"
	"cwtch.im/cwtch/model"
	"cwtch.im/cwtch/peer"
	"cwtch.im/cwtch/protocol/connections"
	"cwtch.im/cwtch/storage"
	"fmt"
	"git.openprivacy.ca/openprivacy/libricochet-go/identity"

	"git.openprivacy.ca/openprivacy/libricochet-go/connectivity"
	"git.openprivacy.ca/openprivacy/libricochet-go/log"
	"io/ioutil"
	"os"
	"path"
	"sync"
)

type applicationCore struct {
	eventBuses map[string]event.Manager

	acn       connectivity.ACN
	directory string
	mutex     sync.Mutex
}

type applicationPeers struct {
	peers map[string]peer.CwtchPeer
}

type application struct {
	applicationCore
	applicationPeers
	storage map[string]storage.ProfileStore
	engines map[string]connections.Engine
	appBus  event.Manager
}

// Application is a full cwtch peer application. It allows management, usage and storage of multiple peers
type Application interface {
	LoadProfiles(password string)
	CreatePeer(name string, password string)
	LaunchPeers()

	GetPrimaryBus() event.Manager
	GetEventBus(onion string) event.Manager

	ShutdownPeer(string)
	Shutdown()

	GetPeer(onion string) peer.CwtchPeer
	ListPeers() map[string]string
}

// LoadProfileFn is the function signature for a function in an app that loads a profile
type LoadProfileFn func(profile *model.Profile, store storage.ProfileStore)

func newAppCore(acn connectivity.ACN, appDirectory string) *applicationCore {
	appCore := &applicationCore{eventBuses: make(map[string]event.Manager), directory: appDirectory, acn: acn}
	os.MkdirAll(path.Join(appCore.directory, "profiles"), 0700)
	return appCore
}

// NewApp creates a new app with some environment awareness and initializes a Tor Manager
func NewApp(acn connectivity.ACN, appDirectory string) Application {
	log.Debugf("NewApp(%v)\n", appDirectory)
	app := &application{storage: make(map[string]storage.ProfileStore), applicationPeers: applicationPeers{peers: make(map[string]peer.CwtchPeer)}, engines: make(map[string]connections.Engine), applicationCore: *newAppCore(acn, appDirectory), appBus: event.NewEventManager()}
	return app
}

// CreatePeer creates a new Peer with a given name and core required accessories (eventbus)
func (ac *applicationCore) CreatePeer(name string, password string) (*model.Profile, error) {
	log.Debugf("CreatePeer(%v)\n", name)

	profile := storage.NewProfile(name)

	ac.mutex.Lock()
	defer ac.mutex.Unlock()

	_, exists := ac.eventBuses[profile.Onion]
	if exists {
		return nil, fmt.Errorf("Error: profile for onion %v already exists", profile.Onion)
	}

	eventBus := event.NewEventManager()
	ac.eventBuses[profile.Onion] = eventBus

	return profile, nil
}

// CreatePeer creates a new Peer with the given name and required accessories (eventbus, storage, protocol engine)
func (app *application) CreatePeer(name string, password string) {
	profile, err := app.applicationCore.CreatePeer(name, password)
	if err != nil {
		app.appBus.Publish(event.NewEventList(event.PeerError, event.Error, err.Error()))
		return
	}

	profileStore := storage.NewProfileWriterStore(app.eventBuses[profile.Onion], path.Join(app.directory, "profiles", profile.LocalID), password, profile)
	app.storage[profile.Onion] = profileStore

	pc := app.storage[profile.Onion].GetProfileCopy()
	peer := peer.FromProfile(pc)
	peer.Init(app.acn, app.eventBuses[profile.Onion])

	blockedPeers := profile.BlockedPeers()
	// TODO: Would be nice if ProtocolEngine did not need to explicitly be given the Private Key.
	identity := identity.InitializeV3(profile.Name, &profile.Ed25519PrivateKey, &profile.Ed25519PublicKey)
	engine := connections.NewProtocolEngine(identity, profile.Ed25519PrivateKey, app.acn, app.eventBuses[profile.Onion], blockedPeers)

	app.peers[profile.Onion] = peer
	app.engines[profile.Onion] = engine

	app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.Onion}))
}

// LoadProfiles takes a password and attempts to load any profiles it can from storage with it and create Peers for them
func (ac *applicationCore) LoadProfiles(password string, loadProfileFn LoadProfileFn) error {
	files, err := ioutil.ReadDir(path.Join(ac.directory, "profiles"))
	if err != nil {
		return fmt.Errorf("Error: cannot read profiles directory: %v", err)
	}

	for _, file := range files {
		eventBus := event.NewEventManager()
		profileStore := storage.NewProfileWriterStore(eventBus, path.Join(ac.directory, "profiles", file.Name()), password, nil)
		err = profileStore.Load()
		if err != nil {
			continue
		}

		profile := profileStore.GetProfileCopy()

		_, exists := ac.eventBuses[profile.Onion]
		if exists {
			profileStore.Shutdown()
			eventBus.Shutdown()
			log.Errorf("profile for onion %v already exists", profile.Onion)
			continue
		}

		ac.mutex.Lock()
		ac.eventBuses[profile.Onion] = eventBus
		ac.mutex.Unlock()

		loadProfileFn(profile, profileStore)
	}
	return nil
}

// LoadProfiles takes a password and attempts to load any profiles it can from storage with it and create Peers for them
func (app *application) LoadProfiles(password string) {
	app.applicationCore.LoadProfiles(password, func(profile *model.Profile, profileStore storage.ProfileStore) {
		peer := peer.FromProfile(profile)
		peer.Init(app.acn, app.eventBuses[profile.Onion])

		blockedPeers := profile.BlockedPeers()
		identity := identity.InitializeV3(profile.Name, &profile.Ed25519PrivateKey, &profile.Ed25519PublicKey)
		engine := connections.NewProtocolEngine(identity, profile.Ed25519PrivateKey, app.acn, app.eventBuses[profile.Onion], blockedPeers)
		app.mutex.Lock()
		app.peers[profile.Onion] = peer
		app.storage[profile.Onion] = profileStore
		app.engines[profile.Onion] = engine
		app.mutex.Unlock()
		app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.Onion}))
	})
}

// GetPrimaryBus returns the bus the Application uses for events that aren't peer specific
func (app *application) GetPrimaryBus() event.Manager {
	return app.appBus
}

// LaunchPeers starts each peer Listening and connecting to peers and groups
func (appPeers *applicationPeers) LaunchPeers() {
	for _, p := range appPeers.peers {
		if !p.IsStarted() {
			p.Listen()
			p.StartPeersConnections()
			p.StartGroupConnections()
		}
	}
}

// ListPeers returns a map of onions to their profile's Name
func (appPeers *applicationPeers) ListPeers() map[string]string {
	keys := map[string]string{}
	for k, p := range appPeers.peers {
		keys[k] = p.GetProfile().Name
	}
	return keys
}

// GetPeer returns a cwtchPeer for a given onion address
func (appPeers *applicationPeers) GetPeer(onion string) peer.CwtchPeer {
	if peer, ok := appPeers.peers[onion]; ok {
		return peer
	}
	return nil
}

// GetEventBus returns a cwtchPeer's event bus
func (ac *applicationCore) GetEventBus(onion string) event.Manager {
	if manager, ok := ac.eventBuses[onion]; ok {
		return manager
	}
	return nil
}

// ShutdownPeer shuts down a peer and removes it from the app's management
func (app *application) ShutdownPeer(onion string) {
	app.mutex.Lock()
	defer app.mutex.Unlock()
	app.eventBuses[onion].Shutdown()
	delete(app.eventBuses, onion)
	app.peers[onion].Shutdown()
	delete(app.peers, onion)
	app.engines[onion].Shutdown()
	delete(app.engines, onion)
	app.storage[onion].Shutdown()
	delete(app.storage, onion)
}

// Shutdown shutsdown all peers of an app and then the tormanager
func (app *application) Shutdown() {
	for id, peer := range app.peers {
		peer.Shutdown()
		app.engines[id].Shutdown()
		app.storage[id].Shutdown()
		app.eventBuses[id].Shutdown()
	}
	app.appBus.Shutdown()
}