Merge remote-tracking branch 'origin/wip'

4 files changed, 575 insertions, 60 deletions

diff --git a/lib/syncthing/folder-st.go b/lib/syncthing/folder-st.go
new file mode 100644
index 0000000..da27062
--- /dev/null
+++ b/lib/syncthing/folder-st.go
@@ -0,0 +1,170 @@
+package st
+
+import (
+	"fmt"
+	"path/filepath"
+
+	"github.com/iotbzh/xds-server/lib/folder"
+	"github.com/iotbzh/xds-server/lib/xdsconfig"
+	uuid "github.com/satori/go.uuid"
+	"github.com/syncthing/syncthing/lib/config"
+)
+
+// IFOLDER interface implementation for syncthing
+
+// STFolder .
+type STFolder struct {
+	globalConfig      *xdsconfig.Config
+	st                *SyncThing
+	fConfig           folder.FolderConfig
+	stfConfig         config.FolderConfiguration
+	eventIDs          []int
+	eventChangeCB     *folder.EventCB
+	eventChangeCBData *folder.EventCBData
+}
+
+// NewFolderST Create a new instance of STFolder
+func (s *SyncThing) NewFolderST(gc *xdsconfig.Config) *STFolder {
+	return &STFolder{
+		globalConfig: gc,
+		st:           s,
+	}
+}
+
+// NewUID Get a UUID
+func (f *STFolder) NewUID(suffix string) string {
+	i := len(f.st.MyID)
+	if i > 15 {
+		i = 15
+	}
+	return uuid.NewV1().String()[:14] + f.st.MyID[:i] + "_" + suffix
+}
+
+// Add a new folder
+func (f *STFolder) Add(cfg folder.FolderConfig) (*folder.FolderConfig, error) {
+
+	// Sanity check
+	if cfg.DataCloudSync.SyncThingID == "" {
+		return nil, fmt.Errorf("device id not set (SyncThingID field)")
+	}
+
+	// rootPath should not be empty
+	if cfg.RootPath == "" {
+		cfg.RootPath = f.globalConfig.FileConf.ShareRootDir
+	}
+
+	f.fConfig = cfg
+
+	f.fConfig.DataCloudSync.BuilderSThgID = f.st.MyID // FIXME - should be removed after local ST config rework
+
+	// Update Syncthing folder
+	// (expect if status is ErrorConfig)
+	// TODO: add cache to avoid multiple requests on startup
+	if f.fConfig.Status != folder.StatusErrorConfig {
+		id, err := f.st.FolderChange(f.fConfig)
+		if err != nil {
+			return nil, err
+		}
+
+		f.stfConfig, err = f.st.FolderConfigGet(id)
+		if err != nil {
+			f.fConfig.Status = folder.StatusErrorConfig
+			return nil, err
+		}
+
+		// Register to events to update folder status
+		for _, evName := range []string{EventStateChanged, EventFolderPaused} {
+			evID, err := f.st.Events.Register(evName, f.cbEventState, id, nil)
+			if err != nil {
+				return nil, err
+			}
+			f.eventIDs = append(f.eventIDs, evID)
+		}
+
+		f.fConfig.IsInSync = false // will be updated later by events
+		f.fConfig.Status = folder.StatusEnable
+	}
+
+	return &f.fConfig, nil
+}
+
+// GetConfig Get public part of folder config
+func (f *STFolder) GetConfig() folder.FolderConfig {
+	return f.fConfig
+}
+
+// GetFullPath returns the full path
+func (f *STFolder) GetFullPath(dir string) string {
+	if &dir == nil {
+		dir = ""
+	}
+	if filepath.IsAbs(dir) {
+		return filepath.Join(f.fConfig.RootPath, dir)
+	}
+	return filepath.Join(f.fConfig.RootPath, f.fConfig.ClientPath, dir)
+}
+
+// Remove a folder
+func (f *STFolder) Remove() error {
+	return f.st.FolderDelete(f.stfConfig.ID)
+}
+
+// RegisterEventChange requests registration for folder event change
+func (f *STFolder) RegisterEventChange(cb *folder.EventCB, data *folder.EventCBData) error {
+	f.eventChangeCB = cb
+	f.eventChangeCBData = data
+	return nil
+}
+
+// UnRegisterEventChange remove registered callback
+func (f *STFolder) UnRegisterEventChange() error {
+	f.eventChangeCB = nil
+	f.eventChangeCBData = nil
+	return nil
+}
+
+// Sync Force folder files synchronization
+func (f *STFolder) Sync() error {
+	return f.st.FolderScan(f.stfConfig.ID, "")
+}
+
+// IsInSync Check if folder files are in-sync
+func (f *STFolder) IsInSync() (bool, error) {
+	sts, err := f.st.IsFolderInSync(f.stfConfig.ID)
+	if err != nil {
+		return false, err
+	}
+	f.fConfig.IsInSync = sts
+	return sts, nil
+}
+
+// callback use to update IsInSync status
+func (f *STFolder) cbEventState(ev Event, data *EventsCBData) {
+	prevSync := f.fConfig.IsInSync
+	prevStatus := f.fConfig.Status
+
+	switch ev.Type {
+
+	case EventStateChanged:
+		to := ev.Data["to"]
+		switch to {
+		case "scanning", "syncing":
+			f.fConfig.Status = folder.StatusSyncing
+		case "idle":
+			f.fConfig.Status = folder.StatusEnable
+		}
+		f.fConfig.IsInSync = (to == "idle")
+
+	case EventFolderPaused:
+		if f.fConfig.Status == folder.StatusEnable {
+			f.fConfig.Status = folder.StatusPause
+		}
+		f.fConfig.IsInSync = false
+	}
+
+	if f.eventChangeCB != nil &&
+		(prevSync != f.fConfig.IsInSync || prevStatus != f.fConfig.Status) {
+		cpConf := f.fConfig
+		(*f.eventChangeCB)(&cpConf, f.eventChangeCBData)
+	}
+}
diff --git a/lib/syncthing/st.go b/lib/syncthing/st.go
index 3380cda..5086994 100644
--- a/lib/syncthing/st.go
+++ b/lib/syncthing/st.go
@@ -27,11 +27,13 @@ import (
 
 // SyncThing .
 type SyncThing struct {
-	BaseURL string
-	APIKey  string
-	Home    string
-	STCmd   *exec.Cmd
-	STICmd  *exec.Cmd
+	BaseURL   string
+	APIKey    string
+	Home      string
+	STCmd     *exec.Cmd
+	STICmd    *exec.Cmd
+	MyID      string
+	Connected bool
 
 	// Private fields
 	binDir      string
@@ -41,6 +43,7 @@ type SyncThing struct {
 	conf        *xdsconfig.Config
 	client      *common.HTTPClient
 	log         *logrus.Logger
+	Events      *Events
 }
 
 // ExitChan Channel used for process exit
@@ -125,6 +128,9 @@ func NewSyncThing(conf *xdsconfig.Config, log *logrus.Logger) *SyncThing {
 		conf:    conf,
 	}
 
+	// Create Events monitoring
+	s.Events = s.NewEventListener()
+
 	return &s
 }
 
@@ -211,13 +217,13 @@ func (s *SyncThing) Start() (*exec.Cmd, error) {
 	env := []string{
 		"STNODEFAULTFOLDER=1",
 		"STNOUPGRADE=1",
-		"STNORESTART=1",
+		"STNORESTART=1", // FIXME SEB remove ?
 	}
 
 	s.STCmd, err = s.startProc("syncthing", args, env, &s.exitSTChan)
 
 	// Use autogenerated apikey if not set by config.json
-	if s.APIKey == "" {
+	if err == nil && s.APIKey == "" {
 		if fd, err := os.Open(filepath.Join(s.Home, "config.xml")); err == nil {
 			defer fd.Close()
 			if b, err := ioutil.ReadAll(fd); err == nil {
@@ -296,6 +302,7 @@ func (s *SyncThing) StopInotify() {
 // Connect Establish HTTP connection with Syncthing
 func (s *SyncThing) Connect() error {
 	var err error
+	s.Connected = false
 	s.client, err = common.HTTPNewClient(s.BaseURL,
 		common.HTTPClientConfig{
 			URLPrefix:           "/rest",
@@ -312,9 +319,22 @@ func (s *SyncThing) Connect() error {
 		return fmt.Errorf("ERROR: cannot connect to Syncthing (null client)")
 	}
 
-	s.client.SetLogger(s.log)
+	// Redirect HTTP log into a file
+	s.client.SetLogLevel(s.conf.Log.Level.String())
+	s.client.LoggerPrefix = "SYNCTHING: "
+	s.client.LoggerOut = s.conf.LogVerboseOut
+
+	s.MyID, err = s.IDGet()
+	if err != nil {
+		return fmt.Errorf("ERROR: cannot retrieve ID")
+	}
+
+	s.Connected = true
+
+	// Start events monitoring
+	err = s.Events.Start()
 
-	return nil
+	return err
 }
 
 // IDGet returns the Syncthing ID of Syncthing instance running locally
@@ -360,44 +380,3 @@ func (s *SyncThing) IsConfigInSync() (bool, error) {
 	}
 	return d.ConfigInSync, nil
 }
-
-// FolderStatus Returns all information about the current
-func (s *SyncThing) FolderStatus(folderID string) (*FolderStatus, error) {
-	var data []byte
-	var res FolderStatus
-	if folderID == "" {
-		return nil, fmt.Errorf("folderID not set")
-	}
-	if err := s.client.HTTPGet("db/status?folder="+folderID, &data); err != nil {
-		return nil, err
-	}
-	if err := json.Unmarshal(data, &res); err != nil {
-		return nil, err
-	}
-	return &res, nil
-}
-
-// IsFolderInSync Returns true when folder is in sync
-func (s *SyncThing) IsFolderInSync(folderID string) (bool, error) {
-	// FIXME better to detected FolderCompletion event (/rest/events)
-	// See https://docs.syncthing.net/dev/events.html
-	sts, err := s.FolderStatus(folderID)
-	if err != nil {
-		return false, err
-	}
-	return sts.NeedBytes == 0, nil
-}
-
-// FolderScan Request immediate folder scan.
-// Scan all folders if folderID param is empty
-func (s *SyncThing) FolderScan(folderID string, subpath string) error {
-	url := "db/scan"
-	if folderID != "" {
-		url += "?folder=" + folderID
-
-		if subpath != "" {
-			url += "&sub=" + subpath
-		}
-	}
-	return s.client.HTTPPost(url, "")
-}
diff --git a/lib/syncthing/stEvent.go b/lib/syncthing/stEvent.go
new file mode 100644
index 0000000..9ca8b78
--- /dev/null
+++ b/lib/syncthing/stEvent.go
@@ -0,0 +1,265 @@
+package st
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/Sirupsen/logrus"
+)
+
+// Events .
+type Events struct {
+	MonitorTime time.Duration
+	Debug       bool
+
+	stop  chan bool
+	st    *SyncThing
+	log   *logrus.Logger
+	cbArr map[string][]cbMap
+}
+
+type Event struct {
+	Type string            `json:"type"`
+	Time time.Time         `json:"time"`
+	Data map[string]string `json:"data"`
+}
+
+type EventsCBData map[string]interface{}
+type EventsCB func(ev Event, cbData *EventsCBData)
+
+const (
+	EventFolderCompletion string = "FolderCompletion"
+	EventFolderSummary    string = "FolderSummary"
+	EventFolderPaused     string = "FolderPaused"
+	EventFolderResumed    string = "FolderResumed"
+	EventFolderErrors     string = "FolderErrors"
+	EventStateChanged     string = "StateChanged"
+)
+
+var EventsAll string = EventFolderCompletion + "|" +
+	EventFolderSummary + "|" +
+	EventFolderPaused + "|" +
+	EventFolderResumed + "|" +
+	EventFolderErrors + "|" +
+	EventStateChanged
+
+type STEvent struct {
+	// Per-subscription sequential event ID. Named "id" for backwards compatibility with the REST API
+	SubscriptionID int `json:"id"`
+	// Global ID of the event across all subscriptions
+	GlobalID int                    `json:"globalID"`
+	Time     time.Time              `json:"time"`
+	Type     string                 `json:"type"`
+	Data     map[string]interface{} `json:"data"`
+}
+
+type cbMap struct {
+	id       int
+	cb       EventsCB
+	filterID string
+	data     *EventsCBData
+}
+
+// NewEventListener Create a new instance of Event listener
+func (s *SyncThing) NewEventListener() *Events {
+	_, dbg := os.LookupEnv("XDS_DEBUG_STEVENTS") // set to add more debug log
+	return &Events{
+		MonitorTime: 100, // in Milliseconds
+		Debug:       dbg,
+		stop:        make(chan bool, 1),
+		st:          s,
+		log:         s.log,
+		cbArr:       make(map[string][]cbMap),
+	}
+}
+
+// Start starts event monitoring loop
+func (e *Events) Start() error {
+	go e.monitorLoop()
+	return nil
+}
+
+// Stop stops event monitoring loop
+func (e *Events) Stop() {
+	e.stop <- true
+}
+
+// Register Add a listener on an event
+func (e *Events) Register(evName string, cb EventsCB, filterID string, data *EventsCBData) (int, error) {
+	if evName == "" || !strings.Contains(EventsAll, evName) {
+		return -1, fmt.Errorf("Unknown event name")
+	}
+	if data == nil {
+		data = &EventsCBData{}
+	}
+
+	cbList := []cbMap{}
+	if _, ok := e.cbArr[evName]; ok {
+		cbList = e.cbArr[evName]
+	}
+
+	id := len(cbList)
+	(*data)["id"] = strconv.Itoa(id)
+
+	e.cbArr[evName] = append(cbList, cbMap{id: id, cb: cb, filterID: filterID, data: data})
+
+	return id, nil
+}
+
+// UnRegister Remove a listener event
+func (e *Events) UnRegister(evName string, id int) error {
+	cbKey, ok := e.cbArr[evName]
+	if !ok {
+		return fmt.Errorf("No event registered to such name")
+	}
+
+	// FIXME - NOT TESTED
+	if id >= len(cbKey) {
+		return fmt.Errorf("Invalid id")
+	} else if id == len(cbKey) {
+		e.cbArr[evName] = cbKey[:id-1]
+	} else {
+		e.cbArr[evName] = cbKey[id : id+1]
+	}
+
+	return nil
+}
+
+// GetEvents returns the Syncthing events
+func (e *Events) getEvents(since int) ([]STEvent, error) {
+	var data []byte
+	ev := []STEvent{}
+	url := "events"
+	if since != -1 {
+		url += "?since=" + strconv.Itoa(since)
+	}
+	if err := e.st.client.HTTPGet(url, &data); err != nil {
+		return ev, err
+	}
+	err := json.Unmarshal(data, &ev)
+	return ev, err
+}
+
+// Loop to monitor Syncthing events
+func (e *Events) monitorLoop() {
+	e.log.Infof("Event monitoring running...")
+	since := 0
+	cntErrConn := 0
+	cntErrRetry := 1
+	for {
+		select {
+		case <-e.stop:
+			e.log.Infof("Event monitoring exited")
+			return
+
+		case <-time.After(e.MonitorTime * time.Millisecond):
+
+			if !e.st.Connected {
+				cntErrConn++
+				time.Sleep(time.Second)
+				if cntErrConn > cntErrRetry {
+					e.log.Error("ST Event monitor: ST connection down")
+					cntErrConn = 0
+					cntErrRetry *= 2
+					if _, err := e.getEvents(since); err == nil {
+						e.st.Connected = true
+						cntErrRetry = 1
+						// XXX - should we reset since value ?
+						goto readEvent
+					}
+				}
+				continue
+			}
+
+		readEvent:
+			stEvArr, err := e.getEvents(since)
+			if err != nil {
+				e.log.Errorf("Syncthing Get Events: %v", err)
+				e.st.Connected = false
+				continue
+			}
+
+			// Process events
+			for _, stEv := range stEvArr {
+				since = stEv.SubscriptionID
+				if e.Debug {
+					e.log.Warnf("ST EVENT: %d %s\n  %v", stEv.GlobalID, stEv.Type, stEv)
+				}
+
+				cbKey, ok := e.cbArr[stEv.Type]
+				if !ok {
+					continue
+				}
+
+				evData := Event{
+					Type: stEv.Type,
+					Time: stEv.Time,
+				}
+
+				// Decode Events
+				// FIXME: re-define data struct for each events
+				// instead of map of string and use JSON marshing/unmarshing
+				fID := ""
+				evData.Data = make(map[string]string)
+				switch stEv.Type {
+
+				case EventFolderCompletion:
+					fID = convString(stEv.Data["folder"])
+					evData.Data["completion"] = convFloat64(stEv.Data["completion"])
+
+				case EventFolderSummary:
+					fID = convString(stEv.Data["folder"])
+					evData.Data["needBytes"] = convInt64(stEv.Data["needBytes"])
+					evData.Data["state"] = convString(stEv.Data["state"])
+
+				case EventFolderPaused, EventFolderResumed:
+					fID = convString(stEv.Data["id"])
+					evData.Data["label"] = convString(stEv.Data["label"])
+
+				case EventFolderErrors:
+					fID = convString(stEv.Data["folder"])
+					// TODO decode array evData.Data["errors"] = convString(stEv.Data["errors"])
+
+				case EventStateChanged:
+					fID = convString(stEv.Data["folder"])
+					evData.Data["from"] = convString(stEv.Data["from"])
+					evData.Data["to"] = convString(stEv.Data["to"])
+
+				default:
+					e.log.Warnf("Unsupported event type")
+				}
+
+				if fID != "" {
+					evData.Data["id"] = fID
+				}
+
+				// Call all registered callbacks
+				for _, c := range cbKey {
+					if e.Debug {
+						e.log.Warnf("EVENT CB fID=%s, filterID=%s", fID, c.filterID)
+					}
+					// Call when filterID is not set or when it matches
+					if c.filterID == "" || (fID != "" && fID == c.filterID) {
+						c.cb(evData, c.data)
+					}
+				}
+			}
+		}
+	}
+}
+
+func convString(d interface{}) string {
+	return d.(string)
+}
+
+func convFloat64(d interface{}) string {
+	return strconv.FormatFloat(d.(float64), 'f', -1, 64)
+}
+
+func convInt64(d interface{}) string {
+	return strconv.FormatInt(d.(int64), 10)
+}
diff --git a/lib/syncthing/stfolder.go b/lib/syncthing/stfolder.go
index 661e19d..70ac70a 100644
--- a/lib/syncthing/stfolder.go
+++ b/lib/syncthing/stfolder.go
@@ -1,34 +1,77 @@
 package st
 
 import (
+	"encoding/json"
+	"fmt"
 	"path/filepath"
 	"strings"
 
-	"github.com/iotbzh/xds-server/lib/xdsconfig"
+	"github.com/iotbzh/xds-server/lib/folder"
 	"github.com/syncthing/syncthing/lib/config"
 	"github.com/syncthing/syncthing/lib/protocol"
 )
 
+// FolderLoadFromStConfig Load/Retrieve folder config from syncthing database
+func (s *SyncThing) FolderLoadFromStConfig(f *[]folder.FolderConfig) error {
+
+	defaultSdk := "" // cannot know which was the default sdk
+
+	stCfg, err := s.ConfigGet()
+	if err != nil {
+		return err
+	}
+	if len(stCfg.Devices) < 1 {
+		return fmt.Errorf("Cannot load syncthing config: no device defined")
+	}
+	devID := stCfg.Devices[0].DeviceID.String()
+	if devID == s.MyID {
+		if len(stCfg.Devices) < 2 {
+			return fmt.Errorf("Cannot load syncthing config: no valid device found")
+		}
+		devID = stCfg.Devices[1].DeviceID.String()
+	}
+
+	for _, stFld := range stCfg.Folders {
+		cliPath := strings.TrimPrefix(stFld.RawPath, s.conf.FileConf.ShareRootDir)
+		if cliPath == "" {
+			cliPath = stFld.RawPath
+		}
+		*f = append(*f, folder.FolderConfig{
+			ID:            stFld.ID,
+			Label:         stFld.Label,
+			ClientPath:    strings.TrimRight(cliPath, "/"),
+			Type:          folder.TypeCloudSync,
+			Status:        folder.StatusDisable,
+			DefaultSdk:    defaultSdk,
+			RootPath:      s.conf.FileConf.ShareRootDir,
+			DataCloudSync: folder.CloudSyncConfig{SyncThingID: devID},
+		})
+	}
+
+	return nil
+}
+
 // FolderChange is called when configuration has changed
-func (s *SyncThing) FolderChange(f xdsconfig.FolderConfig) error {
+func (s *SyncThing) FolderChange(f folder.FolderConfig) (string, error) {
 
 	// Get current config
 	stCfg, err := s.ConfigGet()
 	if err != nil {
 		s.log.Errorln(err)
-		return err
+		return "", err
 	}
 
+	stClientID := f.DataCloudSync.SyncThingID
 	// Add new Device if needed
 	var devID protocol.DeviceID
-	if err := devID.UnmarshalText([]byte(f.SyncThingID)); err != nil {
-		s.log.Errorf("not a valid device id (err %v)\n", err)
-		return err
+	if err := devID.UnmarshalText([]byte(stClientID)); err != nil {
+		s.log.Errorf("not a valid device id (err %v)", err)
+		return "", err
 	}
 
 	newDevice := config.DeviceConfiguration{
 		DeviceID:  devID,
-		Name:      f.SyncThingID,
+		Name:      stClientID,
 		Addresses: []string{"dynamic"},
 	}
 
@@ -49,13 +92,13 @@ func (s *SyncThing) FolderChange(f xdsconfig.FolderConfig) error {
 		label = strings.Split(id, "/")[0]
 	}
 	if id = f.ID; id == "" {
-		id = f.SyncThingID[0:15] + "_" + label
+		id = stClientID[0:15] + "_" + label
 	}
 
 	folder := config.FolderConfiguration{
 		ID:      id,
 		Label:   label,
-		RawPath: filepath.Join(s.conf.FileConf.ShareRootDir, f.RelativePath),
+		RawPath: filepath.Join(s.conf.FileConf.ShareRootDir, f.ClientPath),
 	}
 
 	if s.conf.FileConf.SThgConf.RescanIntervalS > 0 {
@@ -85,7 +128,7 @@ func (s *SyncThing) FolderChange(f xdsconfig.FolderConfig) error {
 		s.log.Errorln(err)
 	}
 
-	return nil
+	return id, nil
 }
 
 // FolderDelete is called to delete a folder config
@@ -110,3 +153,61 @@ func (s *SyncThing) FolderDelete(id string) error {
 
 	return nil
 }
+
+// FolderConfigGet Returns the configuration of a specific folder
+func (s *SyncThing) FolderConfigGet(folderID string) (config.FolderConfiguration, error) {
+	fc := config.FolderConfiguration{}
+	if folderID == "" {
+		return fc, fmt.Errorf("folderID not set")
+	}
+	cfg, err := s.ConfigGet()
+	if err != nil {
+		return fc, err
+	}
+	for _, f := range cfg.Folders {
+		if f.ID == folderID {
+			fc = f
+			return fc, nil
+		}
+	}
+	return fc, fmt.Errorf("id not found")
+}
+
+// FolderStatus Returns all information about the current
+func (s *SyncThing) FolderStatus(folderID string) (*FolderStatus, error) {
+	var data []byte
+	var res FolderStatus
+	if folderID == "" {
+		return nil, fmt.Errorf("folderID not set")
+	}
+	if err := s.client.HTTPGet("db/status?folder="+folderID, &data); err != nil {
+		return nil, err
+	}
+	if err := json.Unmarshal(data, &res); err != nil {
+		return nil, err
+	}
+	return &res, nil
+}
+
+// IsFolderInSync Returns true when folder is in sync
+func (s *SyncThing) IsFolderInSync(folderID string) (bool, error) {
+	sts, err := s.FolderStatus(folderID)
+	if err != nil {
+		return false, err
+	}
+	return sts.NeedBytes == 0 && sts.State == "idle", nil
+}
+
+// FolderScan Request immediate folder scan.
+// Scan all folders if folderID param is empty
+func (s *SyncThing) FolderScan(folderID string, subpath string) error {
+	url := "db/scan"
+	if folderID != "" {
+		url += "?folder=" + folderID
+
+		if subpath != "" {
+			url += "&sub=" + subpath
+		}
+	}
+	return s.client.HTTPPost(url, "")
+}