Refactor subscription checking, add interval checks (1/24h)

cmd/crm/main.go

@@ -4,20 +4,34 @@ import (
 	"log"
 	"os"
 
-	service "github.com/crusttech/crust/crm"
+	context "github.com/SentimensRG/ctx"
+	"github.com/SentimensRG/ctx/sigctx"
 
+	sub "github.com/crusttech/crust/crm"
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/rbac"
+	"github.com/crusttech/crust/internal/subscription"
+	"github.com/crusttech/crust/internal/version"
 )
 
 func main() {
-	flags("crm", service.Flags, auth.Flags, rbac.Flags)
+	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
+
+	ctx := context.AsContext(sigctx.New())
+
+	flags(
+		"crm",
+		sub.Flags,
+		auth.Flags,
+		rbac.Flags,
+		subscription.Flags,
+	)
 
 	// log to stdout not stderr
 	log.SetOutput(os.Stdout)
 	log.SetFlags(log.LstdFlags | log.Lshortfile)
 
-	if err := service.Init(); err != nil {
+	if err := sub.Init(); err != nil {
 		log.Fatalf("Error initializing: %+v", err)
 	}
 
@@ -28,9 +42,12 @@ func main() {
 
 	switch command {
 	case "help":
-	case "merge-users":
 	default:
-		if err := service.Start(); err != nil {
+		// Checks subscription, will os.Exit(1) if there is an error
+		// Disabled for now, system service is the only one that validates subscription
+		// ctx = subscription.Monitor(ctx)
+
+		if err := sub.StartRestAPI(ctx); err != nil {
 			log.Fatalf("Error starting/running: %+v", err)
 		}
 	}

cmd/messaging/main.go

@@ -4,20 +4,34 @@ import (
 	"log"
 	"os"
 
-	service "github.com/crusttech/crust/messaging"
+	context "github.com/SentimensRG/ctx"
+	"github.com/SentimensRG/ctx/sigctx"
 
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/rbac"
+	"github.com/crusttech/crust/internal/subscription"
+	"github.com/crusttech/crust/internal/version"
+	sub "github.com/crusttech/crust/messaging"
 )
 
 func main() {
-	flags("messaging", service.Flags, auth.Flags, rbac.Flags)
+	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
+
+	ctx := context.AsContext(sigctx.New())
+
+	flags(
+		"messaging",
+		sub.Flags,
+		auth.Flags,
+		rbac.Flags,
+		subscription.Flags,
+	)
 
 	// log to stdout not stderr
 	log.SetOutput(os.Stdout)
 	log.SetFlags(log.LstdFlags | log.Lshortfile)
 
-	if err := service.Init(); err != nil {
+	if err := sub.Init(); err != nil {
 		log.Fatalf("Error initializing: %+v", err)
 	}
 
@@ -30,7 +44,11 @@ func main() {
 	case "help":
 	case "merge-users":
 	default:
-		if err := service.Start(); err != nil {
+		// Checks subscription, will os.Exit(1) if there is an error
+		// Disabled for now, system service is the only one that validates subscription
+		// ctx = subscription.Monitor(ctx)
+
+		if err := sub.StartRestAPI(ctx); err != nil {
 			log.Fatalf("Error starting/running: %+v", err)
 		}
 	}

cmd/system/main.go

@@ -7,19 +7,21 @@ import (
 	context "github.com/SentimensRG/ctx"
 	"github.com/SentimensRG/ctx/sigctx"
 
-	"github.com/crusttech/crust/internal/subscription"
-	service "github.com/crusttech/crust/system"
-
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/rbac"
+	"github.com/crusttech/crust/internal/subscription"
+	"github.com/crusttech/crust/internal/version"
+	sub "github.com/crusttech/crust/system"
 )
 
 func main() {
+	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
+
 	ctx := context.AsContext(sigctx.New())
 
 	flags(
 		"system",
-		service.Flags,
+		sub.Flags,
 		auth.Flags,
 		rbac.Flags,
 		subscription.Flags,
@@ -29,7 +31,7 @@ func main() {
 	log.SetOutput(os.Stdout)
 	log.SetFlags(log.LstdFlags | log.Lshortfile)
 
-	if err := service.Init(); err != nil {
+	if err := sub.Init(); err != nil {
 		log.Fatalf("Error initializing: %+v", err)
 	}
 
@@ -40,18 +42,11 @@ func main() {
 
 	switch command {
 	case "help":
-	case "merge-users":
 	default:
-		log.Println("Validating subscription")
-		// Checks subscription & runs internal checker that runs every 24h
-		if err := subscription.Check(ctx); err != nil {
-			log.Printf("Subscription could not be validated, reason: %v", err)
-			os.Exit(-1)
-		} else {
-			log.Println("Subscription valdiated")
-		}
+		// Checks subscription, will os.Exit(1) if there is an error
+		ctx = subscription.Monitor(ctx)
 
-		if err := service.StartRestAPI(ctx); err != nil {
+		if err := sub.StartRestAPI(ctx); err != nil {
 			log.Fatalf("Error starting/running: %+v", err)
 		}
 	}

crm/start.go

@@ -1,14 +1,12 @@
 package service
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"net"
 	"net/http"
-	"os"
 
-	"github.com/SentimensRG/ctx"
-	"github.com/SentimensRG/ctx/sigctx"
 	"github.com/pkg/errors"
 	"github.com/titpetric/factory/resputil"
 
@@ -20,7 +18,6 @@ import (
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/mail"
 	"github.com/crusttech/crust/internal/metrics"
-	"github.com/crusttech/crust/internal/version"
 )
 
 var (
@@ -71,8 +68,7 @@ func Init() error {
 	return nil
 }
 
-func Start() error {
-	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
+func StartRestAPI(ctx context.Context) error {
 	log.Println("Starting http server on address " + flags.http.Addr)
 	listener, err := net.Listen("tcp", flags.http.Addr)
 	if err != nil {
@@ -83,9 +79,8 @@ func Start() error {
 		go metrics.NewMonitor(flags.monitor.Interval)
 	}
 
-	var deadline = sigctx.New()
-	go http.Serve(listener, Routes(ctx.AsContext(deadline)))
-	<-deadline.Done()
+	go http.Serve(listener, Routes(ctx))
+	<-ctx.Done()
 
 	return nil
 }

internal/subscription/check.go

@@ -2,23 +2,41 @@ package subscription
 
 import (
 	"context"
+	"log"
+	"time"
 
 	"github.com/pkg/errors"
 
 	"github.com/crusttech/permit/pkg/permit"
 )
 
+const (
+	// How many times we try to get the licence
+	checkMaxTries = 10
+
+	// Number of seconds between each try (x number of tries)
+	checkTryDelay = 10
+
+	// Timeout in seconds
+	checkTimeout = 30
+)
+
 // Check for subscription
-func Check(ctx context.Context) error {
-	p := permit.Permit{
-		Key:    flags.subscription.Key,
-		Domain: flags.subscription.Domain,
-	}
+func Check(ctx context.Context) (err error) {
+	var (
+		done context.CancelFunc
+		p    *permit.Permit
+		try  = 1
+		in   = permit.Permit{
+			Key:    flags.subscription.Key,
+			Domain: flags.subscription.Domain,
+		}
+	)
 
 	// Do not collect stats on local domains.
 	// if p.Domain != "local.crust.tech" {
 	// @todo collect & pass attributes (no of users....) to be validated by permit.crust.tech subscription server.
-	p.Attributes = map[string]int{
+	in.Attributes = map[string]int{
 		"messaging.enabled": 1,
 		// "messaging.max-public-channels": 1,
 		// "messaging.max-messages": 1,
@@ -40,9 +58,29 @@ func Check(ctx context.Context) error {
 	}
 	// }
 
-	if p, err := permit.Check(ctx, p); err != nil {
-		return errors.Wrap(err, "unable to check for licence")
-	} else if !p.IsValid() {
+	for {
+		ctx, done = context.WithTimeout(ctx, time.Second*checkTimeout)
+		defer done()
+
+		p, err = permit.Check(ctx, in)
+		if err == nil {
+			break
+		}
+
+		if err != nil {
+			log.Println("unable to check for licence: %v, try %d/%d", err, try, checkMaxTries)
+		}
+
+		if try >= checkMaxTries {
+			return errors.Wrap(err, "unable to check for licence")
+		}
+
+		time.Sleep(time.Second * checkTryDelay * checkTryDelay)
+
+		try++
+	}
+
+	if !p.IsValid() {
 		return err
 	} else if p.Domain != flags.subscription.Domain {
 		return errors.Errorf("subscription domains do not match (%s <> %s)", p.Domain, flags.subscription.Domain)

internal/subscription/monitor.go

@@ -0,0 +1,54 @@
+package subscription
+
+import (
+	"context"
+	"log"
+	"os"
+	"time"
+)
+
+// Starts subscription checker
+func Monitor(ctx context.Context) context.Context {
+	check := func(ctx context.Context) bool {
+		log.Println("Validating subscription")
+		if err := Check(ctx); err != nil {
+			log.Printf("Subscription could not be validated, reason: %v", err)
+			return false
+		} else {
+			log.Println("Subscription validated")
+		}
+
+		return true
+	}
+
+	if !check(ctx) {
+		// Initial subscription check failed,
+		// Just exit.
+		os.Exit(1)
+	}
+
+	// Initialize new context with cancellation we'll return this context and use it from this point on so that we make
+	// a clean exist in case subscription checking fails
+	ctx, cancel := context.WithCancel(ctx)
+
+	go func() {
+		// Validate subscription key every 24h ours (from the last start of the service)
+		t := time.NewTicker(time.Hour * 24)
+		defer t.Stop()
+
+		for {
+			select {
+			case <-t.C:
+				// Check the subscription again and call cancel on context
+				if !check(ctx) {
+					cancel()
+					os.Exit(1)
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return ctx
+}

messaging/start.go

@@ -1,14 +1,12 @@
 package service
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"net"
 	"net/http"
-	"os"
 
-	"github.com/SentimensRG/ctx"
-	"github.com/SentimensRG/ctx/sigctx"
 	"github.com/pkg/errors"
 	"github.com/titpetric/factory/resputil"
 
@@ -19,7 +17,6 @@ import (
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/mail"
 	"github.com/crusttech/crust/internal/metrics"
-	"github.com/crusttech/crust/internal/version"
 	"github.com/crusttech/crust/messaging/service"
 )
 
@@ -71,8 +68,7 @@ func Init() error {
 	return nil
 }
 
-func Start() error {
-	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
+func StartRestAPI(ctx context.Context) error {
 	log.Println("Starting http server on address " + flags.http.Addr)
 	listener, err := net.Listen("tcp", flags.http.Addr)
 	if err != nil {
@@ -83,9 +79,8 @@ func Start() error {
 		go metrics.NewMonitor(flags.monitor.Interval)
 	}
 
-	var deadline = sigctx.New()
-	go http.Serve(listener, Routes(ctx.AsContext(deadline)))
-	<-deadline.Done()
+	go http.Serve(listener, Routes(ctx))
+	<-ctx.Done()
 
 	return nil
 }

system/start.go

@@ -6,7 +6,6 @@ import (
 	"log"
 	"net"
 	"net/http"
-	"os"
 
 	"github.com/pkg/errors"
 	"github.com/titpetric/factory/resputil"
@@ -18,7 +17,6 @@ import (
 	"github.com/crusttech/crust/internal/auth"
 	"github.com/crusttech/crust/internal/mail"
 	"github.com/crusttech/crust/internal/metrics"
-	"github.com/crusttech/crust/internal/version"
 )
 
 var (
@@ -75,7 +73,6 @@ func InitDb() error {
 }
 
 func StartRestAPI(ctx context.Context) error {
-	log.Printf("Starting "+os.Args[0]+", version: %v, built on: %v", version.Version, version.BuildTime)
 	log.Println("Starting http server on address " + flags.http.Addr)
 	listener, err := net.Listen("tcp", flags.http.Addr)
 	if err != nil {