count users (#679)

* count users

* fix: Add users-count handler and update response. Update Makefile and create onboarding worker air.toml

* fix phony

* refactor

---------

Co-authored-by: stefanskoricdev <stephanskoric@gmail.com>

Author: davidabram

Makefile

@@ -3,10 +3,16 @@ watch-oss-api:
 	@export $$(cat .env | xargs) && \
     ./bin/air -c oss-api.air.toml
 
+.PHONY: watch-internal-api
 watch-internal-api:
 	@export $$(cat .env | xargs) && \
     ./bin/air -c internal-api.air.toml
 
+.PHONY: watch-onboarding-worker
+watch-onboarding-worker:
+	@export $$(cat .env | xargs) && \
+	./bin/air -c onboarding-worker.air.toml
+
 .ONESHELL:
 setup:
 	@curl -sSfL https://raw.githubusercontent.com/cosmtrek/air/master/install.sh | sh -s

cmd/internal-api/main.go

@@ -10,10 +10,12 @@ import (
 	"time"
 
 	"github.com/dxta-dev/app/internal/internal_api/handler"
+	"github.com/dxta-dev/app/internal/onboarding"
 	"github.com/dxta-dev/app/internal/util"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
 	"github.com/go-chi/jwtauth/v5"
+	"go.temporal.io/sdk/client"
 
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	instrruntime "go.opentelemetry.io/contrib/instrumentation/runtime"
@@ -50,6 +52,11 @@ func initTracer(ctx context.Context, res *sdkresource.Resource) (*sdktrace.Trace
 }
 
 func main() {
+	cfg, err := onboarding.LoadConfig()
+	if err != nil {
+		log.Fatalln("Failed to load configuration:", err)
+	}
+
 	isEndpointProvided := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != "" ||
 		os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") != ""
 
@@ -112,6 +119,18 @@ func main() {
 		srv.IdleTimeout = 30 * time.Second
 	}
 
+	temporalClient, err := client.Dial(client.Options{
+		HostPort:  cfg.TemporalHostPort,
+		Namespace: cfg.TemporalOnboardingNamespace,
+	})
+	if err != nil {
+		log.Fatalf("Unable to create Temporal client: %v", err)
+	}
+
+	defer temporalClient.Close()
+
+	usersHandler := handler.NewUsers(temporalClient, *cfg)
+
 	r.Route("/tenant", func(r chi.Router) {
 		if os.Getenv("ENABLE_JWT_AUTH") == "true" {
 			pubKey, _ := util.GetRawPublicKey()
@@ -134,6 +153,12 @@ func main() {
 		w.Write([]byte(`OK`))
 	})
 
+	r.Get("/ready", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(`OK`))
+	})
+
+	r.Get("/users-count", usersHandler.UsersCount)
+
 	go func() {
 		log.Printf("Listening on %s\n", srv.Addr)
 		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {

cmd/onboarding-worker/main.go

@@ -19,19 +19,24 @@ func main() {
 
 	temporalClient, err := client.Dial(client.Options{
 		HostPort:  cfg.TemporalHostPort,
-		Namespace: cfg.TemporalNamespace,
+		Namespace: cfg.TemporalOnboardingNamespace,
 	})
 	if err != nil {
 		log.Fatalln("Unable to create Temporal client", err)
 	}
 	defer temporalClient.Close()
 
-	err = onboarding.RegisterNamespace(context.Background(), cfg.TemporalHostPort, cfg.TemporalNamespace, 30)
+	err = onboarding.RegisterNamespace(
+		context.Background(),
+		cfg.TemporalHostPort,
+		cfg.TemporalOnboardingNamespace,
+		30,
+	)
 	if err != nil {
 		log.Fatalln("Failed to register Temporal namespace:", err)
 	}
 
-	w := worker.New(temporalClient, cfg.TemporalQueueName, worker.Options{})
+	w := worker.New(temporalClient, cfg.TemporalOnboardingQueueName, worker.Options{})
 
 	w.RegisterWorkflow(workflow.CountUsersWorkflow)
 	w.RegisterActivity(activity.CountUsersActivity)

cmd/oss-api/main.go

@@ -154,6 +154,14 @@ func main() {
 	r.Get("/time-to-merge/{org}/{repo}", handler.TimeToMergeHandler)
 	r.Get("/small-mrs/{org}/{repo}", handler.SmallMRsHandler)
 
+	r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(`OK`))
+	})
+
+	r.Get("/ready", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(`OK`))
+	})
+
 	go func() {
 		log.Printf("Listening on %s\n", srv.Addr)
 		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {

internal/internal_api/handler/users_count.go

@@ -0,0 +1,49 @@
+package handler
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/dxta-dev/app/internal/onboarding"
+	"github.com/dxta-dev/app/internal/onboarding/workflow"
+	"github.com/dxta-dev/app/internal/util"
+	"go.temporal.io/sdk/client"
+)
+
+type UsersCountResponse struct {
+	Count int `json:"count"`
+}
+
+type Users struct {
+	temporalClient client.Client
+	config         onboarding.Config
+}
+
+func NewUsers(temporalClient client.Client, config onboarding.Config) *Users {
+	return &Users{
+		temporalClient: temporalClient,
+		config:         config,
+	}
+}
+
+func (u *Users) UsersCount(w http.ResponseWriter, r *http.Request) {
+	out, err := workflow.ExecuteCountUsersWorkflow(r.Context(), u.temporalClient, u.config)
+	if err != nil {
+		log.Fatal(errors.Unwrap(err))
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	if err := json.NewEncoder(w).Encode(UsersCountResponse{Count: out}); err != nil {
+		fmt.Printf("Issue while formatting response. Error: %s", err.Error())
+		util.JSONError(
+			w,
+			util.ErrorParam{Error: "Internal Server Error"},
+			http.StatusInternalServerError,
+		)
+		return
+	}
+}

internal/onboarding/config.go

@@ -7,35 +7,35 @@ import (
 )
 
 type Config struct {
-	TemporalHostPort  string
-	TemporalNamespace string
-	TemporalQueueName string
-	UsersDSN          string
+	TemporalHostPort            string
+	TemporalOnboardingNamespace string
+	TemporalOnboardingQueueName string
+	UsersDSN                    string
 }
 
 func LoadConfig() (*Config, error) {
-	var hostport, namespace, taskQueue, usersDSN string
+	var hostport, onboardingNamespace, onboardingTaskQueue, usersDSN string
 
 	if hostport = os.Getenv("TEMPORAL_HOSTPORT"); hostport == "" {
 		log.Println("TEMPORAL_HOSTPORT not set; using default")
 	}
 
-	if namespace = os.Getenv("TEMPORAL_NAMESPACE"); namespace == "" {
-		return nil, errors.New("TEMPORAL_NAMESPACE is not defined")
+	if onboardingNamespace = os.Getenv("TEMPORAL_ONBOARDING_NAMESPACE"); onboardingNamespace == "" {
+		return nil, errors.New("TEMPORAL_ONBOARDING_NAMESPACE is not defined")
 	}
 
-	if taskQueue = os.Getenv("TEMPORAL_TASK_QUEUE"); taskQueue == "" {
-		return nil, errors.New("TEMPORAL_TASK_QUEUE is not defined")
+	if onboardingTaskQueue = os.Getenv("TEMPORAL_ONBOARDING_TASK_QUEUE"); onboardingTaskQueue == "" {
+		return nil, errors.New("TEMPORAL_ONBOARDING_TASK_QUEUE is not defined")
 	}
 
 	if usersDSN = os.Getenv("USERS_DSN"); usersDSN == "" {
 		return nil, errors.New("USERS_DSN is not defined")
 	}
 
 	return &Config{
-		TemporalHostPort:  hostport,
-		TemporalNamespace: namespace,
-		TemporalQueueName: taskQueue,
-		UsersDSN:          usersDSN,
+		TemporalHostPort:            hostport,
+		TemporalOnboardingNamespace: onboardingNamespace,
+		TemporalOnboardingQueueName: onboardingTaskQueue,
+		UsersDSN:                    usersDSN,
 	}, nil
 }

internal/onboarding/namespace.go

@@ -1,28 +1,24 @@
 package onboarding
 
 import (
-    "context"
-    "fmt"
-    "time"
+	"context"
+	"fmt"
+	"time"
 
-    "google.golang.org/grpc/codes"
-    "google.golang.org/grpc/status"
-    "google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/durationpb"
 
-    "go.temporal.io/api/workflowservice/v1"
-    "go.temporal.io/sdk/client"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/sdk/client"
 )
 
 func RegisterNamespace(ctx context.Context, hostPort, namespace string, retentionDays int) error {
-    nsClient, err := client.NewNamespaceClient(client.Options{HostPort: hostPort})
-    if err != nil {
-        return fmt.Errorf("unable to create namespace client: %w", err)
-    }
-    defer nsClient.Close()
+	nsClient, err := client.NewNamespaceClient(client.Options{HostPort: hostPort})
+	if err != nil {
+		return fmt.Errorf("unable to create namespace client: %w", err)
+	}
+	defer nsClient.Close()
 	if _, err := nsClient.Describe(ctx, namespace); err == nil {
 		return nil
-	} else if s, ok := status.FromError(err); !ok || s.Code() != codes.NotFound {
-		return fmt.Errorf("failed to describe namespace %q: %w", namespace, err)
 	}
 
 	if retentionDays < 1 {
@@ -40,5 +36,5 @@ func RegisterNamespace(ctx context.Context, hostPort, namespace string, retentio
 		return fmt.Errorf("failed to register namespace %q: %w", namespace, err)
 	}
 
-    return nil
+	return nil
 }

internal/onboarding/workflow/count-users.go

@@ -13,7 +13,9 @@ import (
 )
 
 func CountUsersWorkflow(ctx workflow.Context, dsn string) (int, error) {
-	ao := workflow.ActivityOptions{}
+	ao := workflow.ActivityOptions{
+		StartToCloseTimeout: time.Minute * 5,
+	}
 
 	ctx = workflow.WithActivityOptions(ctx, ao)
 
@@ -25,17 +27,29 @@ func CountUsersWorkflow(ctx workflow.Context, dsn string) (int, error) {
 	return count, nil
 }
 
-func ExecuteCountUsersWorkflow(ctx context.Context, temporalClient client.Client, cfg onboarding.Config) (int, error) {
-	wr, err := temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
-		ID:        fmt.Sprintf("count-users-workflow-%v", time.Now().Format("20060102150405")),
-		TaskQueue: cfg.TemporalQueueName,
-	}, CountUsersWorkflow, cfg.UsersDSN)
+func ExecuteCountUsersWorkflow(
+	ctx context.Context,
+	temporalClient client.Client,
+	cfg onboarding.Config,
+) (int, error) {
+	wr, err := temporalClient.ExecuteWorkflow(
+		ctx,
+		client.StartWorkflowOptions{
+			ID:        fmt.Sprintf("count-users-workflow-%v", time.Now().Format("20060102150405")),
+			TaskQueue: cfg.TemporalOnboardingQueueName,
+		},
+		CountUsersWorkflow,
+		cfg.UsersDSN,
+	)
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf("failed to start CountUsersWorkflow: %w", err)
 	}
 
 	var result int
-	wr.Get(ctx, &result)
+	err = wr.Get(ctx, &result)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get result from CountUsersWorkflow: %w", err)
+	}
 
 	return result, nil
 }

onboarding-worker.air.toml

@@ -0,0 +1,51 @@
+root = "."
+testdata_dir = "testdata"
+tmp_dir = "tmp"
+
+[build]
+  args_bin = []
+  bin = "./tmp/onboarding-worker"
+  cmd = "go build -o ./tmp/onboarding-worker ./cmd/onboarding-worker/main.go"
+  delay = 1000
+  exclude_dir = ["assets", "tmp", "vendor", "testdata"]
+  exclude_file = []
+  exclude_regex = ["_test.go"]
+  exclude_unchanged = false
+  follow_symlink = false
+  full_bin = ""
+  include_dir = []
+  include_ext = ["go"]
+  include_file = []
+  kill_delay = "0s"
+  log = "onboarding-worker.build-errors.log"
+  poll = false
+  poll_interval = 0
+  post_cmd = []
+  pre_cmd = []
+  rerun = false
+  rerun_delay = 500
+  send_interrupt = false
+  stop_on_error = false
+
+[color]
+  app = ""
+  build = "yellow"
+  main = "magenta"
+  runner = "green"
+  watcher = "cyan"
+
+[log]
+  main_only = false
+  time = false
+
+[misc]
+  clean_on_exit = false
+
+[proxy]
+  app_port = 0
+  enabled = false
+  proxy_port = 0
+
+[screen]
+  clear_on_rebuild = false
+  keep_scroll = true