chore: dvserver package

Makefile

@@ -49,10 +49,13 @@ gen.sum: $(GEN_SRC)
 	)
 
 
+PROTOC_OPTS = -I ./vendor/github.com/grpc-ecosystem/grpc-gateway:./api:./vendor:/protobuf
 .PHONY: generate_local
 generate_local:
 	@set -e; for proto in $(PROTOS_SRC); do ( set -xe; \
-	  protoc -I ./api:./vendor:/protobuf --grpc-gateway_out=logtostderr=true:"$(GOPATH)/src" --gogofaster_out="plugins=grpc:$(GOPATH)/src" "$$proto" \
+	  protoc $(PROTOC_OPTS) \
+	    --grpc-gateway_out=logtostderr=true:"$(GOPATH)/src" \
+	    --gogofaster_out="plugins=grpc:$(GOPATH)/src" "$$proto" \
 	); done
 	goimports -w ./pkg ./cmd ./internal
 	shasum $(GEN_SRC) | sort > gen.sum.tmp

api/dvmodel.proto

@@ -82,8 +82,8 @@ message Task {
   }
 
   string id = 1 [(gogoproto.casttype) = "github.com/cayleygraph/quad.IRI", (gogoproto.moretags) = "quad:\"@id\"", (gogoproto.customname) = "ID"]; // canonical URI
-  google.protobuf.Timestamp created_at = 3 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false, (gogoproto.moretags) = "quad:\"schema:createdAt,optional\""];
-  google.protobuf.Timestamp updated_at = 4 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false, (gogoproto.moretags) = "quad:\"schema:updatedAt,optional\""];
+  google.protobuf.Timestamp created_at = 3 [(gogoproto.stdtime) = true, (gogoproto.nullable) = true, (gogoproto.moretags) = "quad:\"schema:createdAt,optional\""];
+  google.protobuf.Timestamp updated_at = 4 [(gogoproto.stdtime) = true, (gogoproto.nullable) = true, (gogoproto.moretags) = "quad:\"schema:updatedAt,optional\""];
   string local_id = 5 [(gogoproto.customname) = "LocalID", (gogoproto.moretags) = "quad:\"schema:localId,optional\""];
 
   Kind kind = 10 [(gogoproto.moretags) = "quad:\"schema:kind,optional\""];
@@ -148,3 +148,13 @@ enum Driver {
   Trello = 3;
   Jira = 4;
 }
+
+//
+// internal
+//
+
+message Batch {
+  repeated Task tasks = 1;
+  repeated Owner owners = 2;
+  repeated Topic topics = 3;
+}

api/dvserver.proto

@@ -0,0 +1,39 @@
+syntax = "proto3";
+
+package depviz.server;
+
+import "google/api/annotations.proto";
+import "github.com/gogo/protobuf/gogoproto/gogo.proto";
+import "github.com/grpc-ecosystem/grpc-gateway/protoc-gen-swagger/options/annotations.proto";
+
+import "dvmodel.proto";
+
+option go_package = "moul.io/depviz/internal/dvserver";
+option (gogoproto.sizer_all) = true;
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+
+service DepvizService {
+  rpc Graph(Graph.Input) returns (Graph.Output) { option (google.api.http) = {get: "/graph"}; };
+  rpc StoreDump(StoreDump.Input) returns (StoreDump.Output) { option (google.api.http) = {get: "/store/dump"}; };
+}
+
+message Graph {
+  message Input {
+    repeated string targets = 1;
+    bool with_closed = 2;
+    bool without_isolated = 3;
+    bool without_prs = 4 [(gogoproto.customname) = "WithoutPRs"];
+    bool without_external_deps = 5;
+  }
+  message Output {
+    repeated depviz.model.Task tasks = 1;
+  }
+}
+
+message StoreDump {
+  message Input {}
+  message Output {
+    depviz.model.Batch batch = 1;
+  }
+}

cmd/depviz/main.go

@@ -1,24 +1,29 @@
 package main // import "moul.io/depviz"
 
 import (
+	"context"
 	"errors"
 	"flag"
 	"fmt"
 	"log"
 	"math/rand"
 	"os"
+	"os/signal"
 	"time"
 
 	"github.com/cayleygraph/cayley"
 	"github.com/cayleygraph/cayley/graph"
 	_ "github.com/cayleygraph/cayley/graph/kv/bolt"
 	"github.com/cayleygraph/cayley/schema"
+	"github.com/oklog/run"
 	"github.com/peterbourgon/ff"
 	"github.com/peterbourgon/ff/ffcli"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"moul.io/depviz/internal/dvcore"
+	"moul.io/depviz/internal/dvserver"
 	"moul.io/depviz/internal/dvstore"
+	"moul.io/godev"
 )
 
 var (
@@ -36,9 +41,15 @@ var (
 	airtableTasksTab  = airtableFlags.String("tasks", "Tasks", `"Tasks" tab name`)
 	airtableTopicsTab = airtableFlags.String("topics", "Topics", `"Topics" tab name`)
 
-	serverFlags   = flag.NewFlagSet("server", flag.ExitOnError)
-	serverBind    = serverFlags.String("bind", ":8000", "server bind address")
-	serverGodmode = serverFlags.Bool("godmode", false, "enable dangerous API calls")
+	serverFlags              = flag.NewFlagSet("server", flag.ExitOnError)
+	serverHTTPBind           = serverFlags.String("http-bind", ":8000", "HTTP bind address")
+	serverGRPCBInd           = serverFlags.String("grpc-bind", ":9000", "gRPC bind address")
+	serverRequestTimeout     = serverFlags.Duration("request-timeout", 5*time.Second, "request timeout")
+	serverShutdownTimeout    = serverFlags.Duration("shutdowm-timeout", 6*time.Second, "shutdown timeout")
+	serverCORSAllowedOrigins = serverFlags.String("cors-allowed-origins", "*", "allowed CORS origins")
+	serverGodmode            = serverFlags.Bool("godmode", false, "enable dangerous API calls")
+	serverWithPprof          = serverFlags.Bool("with-pprof", false, "enable pprof endpoints")
+	serverWithoutRecovery    = serverFlags.Bool("without-recovery", false, "disable panic recovery (dev)")
 
 	runFlags            = flag.NewFlagSet("run", flag.ExitOnError)
 	runNoPull           = runFlags.Bool("no-pull", false, "don't pull providers (graph only)")
@@ -201,7 +212,14 @@ func execStoreDumpJSON(args []string) error {
 		return fmt.Errorf("init store: %w", err)
 	}
 
-	return dvcore.StoreDumpJSON(store, schemaConfig)
+	ctx := context.Background()
+	batch, err := dvcore.GetStoreDump(ctx, store, schemaConfig)
+	if err != nil {
+		return fmt.Errorf("get store dump: %w", err)
+	}
+
+	fmt.Println(godev.PrettyJSON(batch))
+	return nil
 }
 
 func execStoreInfo(args []string) error {
@@ -251,12 +269,64 @@ func execServer(args []string) error {
 		return err
 	}
 
-	store, err := storeFromArgs()
-	if err != nil {
-		return fmt.Errorf("init store: %w", err)
+	var (
+		ctx = context.Background()
+		g   run.Group
+		svc dvserver.Service
+	)
+
+	{ // server
+		store, err := storeFromArgs()
+		if err != nil {
+			return fmt.Errorf("init store: %w", err)
+		}
+
+		opts := dvserver.Opts{
+			Logger:             logger,
+			HTTPBind:           *serverHTTPBind,
+			GRPCBind:           *serverGRPCBInd,
+			CORSAllowedOrigins: *serverCORSAllowedOrigins,
+			RequestTimeout:     *serverRequestTimeout,
+			ShutdownTimeout:    *serverShutdownTimeout,
+			WithPprof:          *serverWithPprof,
+			WithoutRecovery:    *serverWithoutRecovery,
+			Godmode:            *serverGodmode,
+		}
+		svc, err = dvserver.New(ctx, store, schemaConfig, opts)
+		if err != nil {
+			return fmt.Errorf("init server: %w", err)
+		}
+
+		g.Add(
+			svc.Run,
+			func(error) { svc.Close() },
+		)
 	}
 
-	return dvcore.Server(*serverBind, *serverGodmode, store, logger, schemaConfig)
+	{ // signal handling
+		ctx, cancel := context.WithCancel(ctx)
+		g.Add(func() error {
+			sigch := make(chan os.Signal, 1)
+			signal.Notify(sigch, os.Interrupt)
+			select {
+			case <-sigch:
+			case <-ctx.Done():
+			}
+			return nil
+		}, func(error) {
+			cancel()
+		})
+	}
+
+	logger.Info("server started",
+		zap.String("http-bind", svc.HTTPListenerAddr()),
+		zap.String("grpc-bind", svc.GRPCListenerAddr()),
+	)
+
+	if err := g.Run(); err != nil {
+		return fmt.Errorf("group terminated: %w", err)
+	}
+	return nil
 }
 
 func storeFromArgs() (*cayley.Handle, error) {

internal/dvcore/run.go

@@ -221,7 +221,7 @@ func saveBatches(h *cayley.Handle, schema *schema.Config, batches []dvmodel.Batc
 				_, _ = schema.WriteAsQuads(dw, working)
 			}
 
-			working = owner
+			working = *owner
 			if _, err := schema.WriteAsQuads(iw, working); err != nil {
 				return fmt.Errorf("write as quads: %w", err)
 			}
@@ -232,7 +232,7 @@ func saveBatches(h *cayley.Handle, schema *schema.Config, batches []dvmodel.Batc
 				_, _ = schema.WriteAsQuads(dw, working)
 			}
 
-			working = task
+			working = *task
 			if _, err := schema.WriteAsQuads(iw, working); err != nil {
 				return fmt.Errorf("write as quads: %w", err)
 			}
@@ -243,7 +243,7 @@ func saveBatches(h *cayley.Handle, schema *schema.Config, batches []dvmodel.Batc
 				_, _ = schema.WriteAsQuads(dw, working)
 			}
 
-			working = topic
+			working = *topic
 			if _, err := schema.WriteAsQuads(iw, working); err != nil {
 				return fmt.Errorf("write as quads: %w", err)
 			}