remote-challenge-sync-airbus: Add inotify watcher

remote/challenge-sync-airbus/main.go

@@ -2,12 +2,14 @@ package main
 
 import (
 	"flag"
+	"io/ioutil"
 	"log"
 	"os"
 	"path"
 	"path/filepath"
 	"strconv"
-	"time"
+
+	"gopkg.in/fsnotify.v1"
 )
 
 var (
@@ -17,9 +19,9 @@ var (
 
 func main() {
 	flag.StringVar(&TeamsDir, "teams", "./TEAMS", "Base directory where save teams JSON files")
-	//var debugINotify = flag.Bool("debuginotify", false, "Show skipped inotofy events")
+	var debugINotify = flag.Bool("debuginotify", false, "Show skipped inotofy events")
 	flag.BoolVar(&skipInitialSync, "skipinitialsync", skipInitialSync, "Skip the initial synchronization")
-	//watcher := flag.Bool("watch", false, "Enable daemon mode by watching the directory")
+	daemon := flag.Bool("watch", false, "Enable daemon mode by watching the directory")
 	tspath := flag.String("timestamp-file", "./REMOTE/timestamp", "Path to the file storing the last timestamp")
 	exercicespath := flag.String("exercices-file", "./REMOTE/exercices-bindings.json", "Path to the file containing the ID bindings")
 	coeff := flag.Float64("global-coeff", 10.0, "Coefficient to use to multiply all scores before passing them to the other platform")
@@ -66,24 +68,94 @@ func main() {
 	}
 
 	w := Walker{
-		LastSync:  *ts,
+		LastSync:  ts,
 		Exercices: exbindings,
 		Teams:     teamsbindings,
 		API:       api,
 		Coeff:     *coeff,
 	}
 
-	// Iterate over teams scores
-	err = filepath.WalkDir(TeamsDir, w.WalkScore)
-	if err != nil {
-		log.Printf("Something goes wrong during walking")
+	if !skipInitialSync {
+		// Iterate over teams scores
+		err = filepath.WalkDir(TeamsDir, w.WalkScore)
+		if err != nil {
+			log.Printf("Something goes wrong during walking")
+		}
+
+		// save current timestamp for teams
+		err = saveTS(*tspath, w.LastSync)
+		if err != nil {
+			log.Fatal("Unable to save timestamp file: ", err.Error())
+		}
 	}
 
-	// Update timestamp for the next time
-	w.LastSync = time.Now()
+	if daemon != nil && *daemon {
+		// Watch teams.json and scores.json
+		log.Println("Registering directory events...")
+		watcher, err := fsnotify.NewWatcher()
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer watcher.Close()
 
-	err = saveTS(*tspath, &w.LastSync)
-	if err != nil {
-		log.Fatal("Unable to save timestamp file: ", err.Error())
+		if err := watchsubdir(watcher, TeamsDir); err != nil {
+			log.Fatal(err)
+		}
+
+		watchedNotify := fsnotify.Create
+
+		for {
+			select {
+			case ev := <-watcher.Events:
+				if d, err := os.Lstat(ev.Name); err == nil && ev.Op&fsnotify.Create == fsnotify.Create && d.Mode().IsDir() && d.Mode()&os.ModeSymlink == 0 && d.Name() != ".tmp" {
+					// Register new subdirectory
+					if err := watchsubdir(watcher, ev.Name); err != nil {
+						log.Println(err)
+					}
+				} else if ev.Op&watchedNotify == watchedNotify && d.Mode().IsRegular() {
+					if *debugINotify {
+						log.Println("Treating event:", ev, "for", ev.Name)
+					}
+					if filepath.Base(ev.Name) == "scores.json" {
+						go w.treat(ev.Name)
+					} else if filepath.Base(ev.Name) == "teams.json" {
+						teamsbindings, err := getTeams(filepath.Join(TeamsDir, "teams.json"))
+						if err != nil {
+							log.Println("Unable to open teams bindings file: ", err.Error())
+							return
+						}
+						w.Teams = teamsbindings
+					}
+				} else if ev.Op&fsnotify.Write == fsnotify.Write {
+					log.Println("FSNOTIFY WRITE SEEN. Prefer looking at them, as it appears files are not atomically moved.")
+					watchedNotify = fsnotify.Write
+				} else if *debugINotify {
+					log.Println("Skipped event:", ev, "for", ev.Name)
+				}
+			case err := <-watcher.Errors:
+				log.Println("error:", err)
+			}
+		}
+	}
+}
+
+func watchsubdir(watcher *fsnotify.Watcher, pathname string) error {
+	log.Println("Watch new directory:", pathname)
+	if err := watcher.Add(pathname); err != nil {
+		return err
+	}
+
+	if ds, err := ioutil.ReadDir(pathname); err != nil {
+		return err
+	} else {
+		for _, d := range ds {
+			p := path.Join(pathname, d.Name())
+			if d.IsDir() && d.Name() != ".tmp" && d.Mode()&os.ModeSymlink == 0 {
+				if err := watchsubdir(watcher, p); err != nil {
+					return err
+				}
+			}
+		}
+		return nil
 	}
 }

remote/challenge-sync-airbus/timestamp.go

@@ -1,50 +1,42 @@
 package main
 
 import (
-	"fmt"
+	"encoding/json"
 	"os"
 	"time"
 )
 
-func loadTS(tspath string) (timestamp *time.Time, err error) {
+func loadTS(tspath string) (timestamp map[AirbusUserId]time.Time, err error) {
+	var fd *os.File
 	if _, err = os.Stat(tspath); os.IsNotExist(err) {
-		init := time.Unix(0, 0)
-		timestamp = &init
-
+		timestamp = map[AirbusUserId]time.Time{}
 		err = saveTS(tspath, timestamp)
 		return
-	}
+	} else if fd, err = os.Open(tspath); err != nil {
+		return nil, err
+	} else {
+		defer fd.Close()
+		jdec := json.NewDecoder(fd)
+
+		if err = jdec.Decode(&timestamp); err != nil {
+			return
+		}
 
-	var fd *os.File
-	fd, err = os.Open(tspath)
-	if err != nil {
 		return
 	}
-	defer fd.Close()
-
-	var init int64
-	_, err = fmt.Fscanf(fd, "%d", &init)
-	if err != nil {
-		return
-	}
-
-	tmp := time.Unix(init, 0)
-	timestamp = &tmp
-
-	return
 }
 
-func saveTS(tspath string, ts *time.Time) error {
-	fd, err := os.Create(tspath)
-	if err != nil {
+func saveTS(tspath string, ts map[AirbusUserId]time.Time) error {
+	if fd, err := os.Create(tspath); err != nil {
 		return err
-	}
-	defer fd.Close()
+	} else {
+		defer fd.Close()
+		jenc := json.NewEncoder(fd)
 
-	_, err = fmt.Fprintf(fd, "%d", ts.Unix())
-	if err != nil {
-		return err
-	}
+		if err := jenc.Encode(ts); err != nil {
+			return err
+		}
 
-	return nil
+		return nil
+	}
 }

remote/challenge-sync-airbus/treat.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"time"
@@ -10,45 +11,52 @@ import (
 )
 
 type Walker struct {
-	LastSync  time.Time
+	LastSync  map[AirbusUserId]time.Time
 	Exercices AirbusExercicesBindings
 	Teams     map[string]fic.ExportedTeam
 	API       AirbusAPI
 	Coeff     float64
 }
 
-func (w *Walker) WalkScore(path string, d os.DirEntry, err error) error {
-	if filepath.Base(path) == "scores.json" {
-		mypath := filepath.Join(filepath.Dir(path), "my.json")
-		if _, err := os.Stat(mypath); !os.IsNotExist(err) {
-			// Read team ID
-			fdmy, err := os.Open(mypath)
-			if err != nil {
-				return err
-			}
-			defer fdmy.Close()
+func (w *Walker) treat(path string) {
+	mypath := filepath.Join(filepath.Dir(path), "my.json")
+	if _, err := os.Stat(mypath); !os.IsNotExist(err) {
+		// Read team ID
+		fdmy, err := os.Open(mypath)
+		if err != nil {
+			log.Println("Unable to open my.json:", err)
+			return
+		}
+		defer fdmy.Close()
 
-			teammy, err := fic.ReadMyJSON(fdmy)
-			if err != nil {
-				return err
-			}
+		teammy, err := fic.ReadMyJSON(fdmy)
+		if err != nil {
+			log.Println("Unable to parse my.json:", err)
+			return
+		}
 
-			airbusTeamId := NewAirbusUserId(w.Teams[fmt.Sprintf("%d", teammy.Id)].ExternalId)
+		airbusTeamId := NewAirbusUserId(w.Teams[fmt.Sprintf("%d", teammy.Id)].ExternalId)
 
-			// Treat score grid
-			err = w.TreatScoreGrid(path, airbusTeamId)
-			if err != nil {
-				return err
-			}
+		// Treat score grid
+		err = w.TreatScoreGrid(path, airbusTeamId)
+		if err != nil {
+			log.Println("Unable to treat score grid:", err)
+			return
+		}
 
-			// Balance scores
-			err = w.BalanceScore(int64(float64(teammy.Points)*w.Coeff), airbusTeamId)
-			if err != nil {
-				return err
-			}
+		// Balance scores
+		err = w.BalanceScore(int64(float64(teammy.Points)*w.Coeff), airbusTeamId)
+		if err != nil {
+			log.Println("Unable to balance score:", err)
+			return
 		}
 	}
+}
 
+func (w *Walker) WalkScore(path string, d os.DirEntry, err error) error {
+	if filepath.Base(path) == "scores.json" {
+		go w.treat(path)
+	}
 	return nil
 }
 
@@ -66,8 +74,12 @@ func (w *Walker) TreatScoreGrid(path string, airbusTeamId AirbusUserId) error {
 	}
 
 	// Found all new entries
+	maxts := w.LastSync[airbusTeamId]
 	for _, row := range teamscores {
-		if row.Time.After(w.LastSync) {
+		if row.Time.After(maxts) {
+			maxts = row.Time
+		}
+		if row.Time.After(w.LastSync[airbusTeamId]) {
 			if row.Reason == "Validation" {
 				err = w.API.ValidateChallengeFromUser(airbusTeamId, w.Exercices[row.IdExercice])
 			} else {
@@ -80,6 +92,8 @@ func (w *Walker) TreatScoreGrid(path string, airbusTeamId AirbusUserId) error {
 		}
 	}
 
+	w.LastSync[airbusTeamId] = maxts
+
 	return nil
 }