From 0719de01c8bbb3bc7ba745db9135b1cffb3ed0f2 Mon Sep 17 00:00:00 2001 From: David Thorpe Date: Fri, 12 Jun 2026 13:12:32 +0200 Subject: [PATCH] Adding pggraph --- cmd/pggraph/main.go | 33 +++++++++++ pggraph/cmd/server.go | 88 +++++++++++++++++++++++++++ pggraph/manager/graph.go | 34 +++++++++++ pggraph/manager/manager.go | 102 ++++++++++++++++++++++++++++++++ pggraph/manager/node.go | 118 +++++++++++++++++++++++++++++++++++++ pggraph/manager/opt.go | 109 ++++++++++++++++++++++++++++++++++ pggraph/manager/run.go | 14 +++++ pggraph/schema/graph.go | 83 ++++++++++++++++++++++++++ pggraph/schema/node.go | 95 +++++++++++++++++++++++++++++ pggraph/schema/objects.sql | 30 ++++++++++ pggraph/schema/queries.sql | 72 ++++++++++++++++++++++ pggraph/schema/schema.go | 18 ++++++ 12 files changed, 796 insertions(+) create mode 100644 cmd/pggraph/main.go create mode 100644 pggraph/cmd/server.go create mode 100644 pggraph/manager/graph.go create mode 100644 pggraph/manager/manager.go create mode 100644 pggraph/manager/node.go create mode 100644 pggraph/manager/opt.go create mode 100644 pggraph/manager/run.go create mode 100644 pggraph/schema/graph.go create mode 100644 pggraph/schema/node.go create mode 100644 pggraph/schema/objects.sql create mode 100644 pggraph/schema/queries.sql create mode 100644 pggraph/schema/schema.go diff --git a/cmd/pggraph/main.go b/cmd/pggraph/main.go new file mode 100644 index 0000000..eedd505 --- /dev/null +++ b/cmd/pggraph/main.go @@ -0,0 +1,33 @@ +package main + +import ( + "fmt" + "os" + + // Packages + pggraphcmd "github.com/mutablelogic/go-pg/pggraph/cmd" + servercmd "github.com/mutablelogic/go-server/pkg/cmd" + version "github.com/mutablelogic/go-server/pkg/version" +) + +/////////////////////////////////////////////////////////////////////////////// +// TYPES + +type CLI struct { + pggraphcmd.ServerCommands // pggraph run +} + +/////////////////////////////////////////////////////////////////////////////// +// GLOBALS + +const description = "PostgreSQL Graph is an application for running DAG graphs" + +/////////////////////////////////////////////////////////////////////////////// +// LIFECYCLE + +func main() { + if err := servercmd.Main(CLI{}, description, version.Version()); err != nil { + fmt.Fprintln(os.Stderr, "Error:", err) + os.Exit(1) + } +} diff --git a/pggraph/cmd/server.go b/pggraph/cmd/server.go new file mode 100644 index 0000000..5a818dd --- /dev/null +++ b/pggraph/cmd/server.go @@ -0,0 +1,88 @@ +package cmd + +import ( + "context" + "fmt" + + // Packages + pg "github.com/mutablelogic/go-pg" + manager "github.com/mutablelogic/go-pg/pggraph/manager" + "github.com/mutablelogic/go-pg/pggraph/schema" + pgpkg "github.com/mutablelogic/go-pg/pkg/cmd" + server "github.com/mutablelogic/go-server" + cmd "github.com/mutablelogic/go-server/pkg/cmd" + "github.com/mutablelogic/go-server/pkg/types" + errgroup "golang.org/x/sync/errgroup" +) + +/////////////////////////////////////////////////////////////////////////////// +// TYPES + +type ServerCommands struct { + Run RunServer `cmd:"" name:"run" help:"Run the server." group:"SERVER"` +} + +type RunServer struct { + cmd.RunServer + pgpkg.PostgresFlags `embed:"" prefix:"pg."` +} + +/////////////////////////////////////////////////////////////////////////////// +// PUBLIC METHODS + +func (runner *RunServer) Run(ctx server.Cmd) error { + // Connect to the database, if configured + conn, err := runner.PostgresFlags.Connect(ctx) + if err != nil { + return err + } else if conn == nil { + return fmt.Errorf("database connection is required") + } + + // Report that the server is running + ctx.Logger().Info("running server", "name", ctx.Name(), "version", ctx.Version()) + + // Create the manager, run the server, and return any error + return runner.WithManager(ctx, conn, func(manager *manager.Manager) error { + // Create an error context - which will cancel any other goroutine on exit + errgroup, errctx := errgroup.WithContext(ctx.Context()) + + // Add a node to the manager + _, err := manager.RegisterNode(ctx.Context(), "helloworld", schema.NodeMeta{ + Description: types.Ptr("Prints a greeting message"), + }, func(ctx context.Context, name string) (string, error) { + return "Hello, " + name, nil + }) + if err != nil { + return err + } + + // Run the manager + errgroup.Go(func() error { + return manager.Run(errctx, ctx.Logger()) + }) + + // Run the server - if any co-routine in the error group returns an error, the server will be shutdown + errgroup.Go(func() error { + return runner.RunServer.Run(ctx.WithContext(errctx)) + }) + + // Wait for the server and manager to exit, and return any error + return errgroup.Wait() + }) +} + +/////////////////////////////////////////////////////////////////////////////// +// PRIVATE METHODS + +func (runner *RunServer) WithManager(ctx server.Cmd, conn pg.PoolConn, fn func(*manager.Manager) error) error { + opts := []manager.Opt{ + manager.WithMeter(ctx.Meter()), + manager.WithTracer(ctx.Tracer()), + } + if manager, err := manager.New(ctx.Context(), conn, ctx.Version(), opts...); err != nil { + return err + } else { + return fn(manager) + } +} diff --git a/pggraph/manager/graph.go b/pggraph/manager/graph.go new file mode 100644 index 0000000..fefad42 --- /dev/null +++ b/pggraph/manager/graph.go @@ -0,0 +1,34 @@ +package manager + +import ( + "context" + + // Packages + otel "github.com/mutablelogic/go-client/pkg/otel" + pg "github.com/mutablelogic/go-pg" + schema "github.com/mutablelogic/go-pg/pggraph/schema" + types "github.com/mutablelogic/go-server/pkg/types" + attribute "go.opentelemetry.io/otel/attribute" +) + +/////////////////////////////////////////////////////////////////////////////// +// PUBLIC METHODS + +// RegisterGraph creates a new graph, or updates an existing graph, and returns it. +func (manager *Manager) RegisterGraph(ctx context.Context, name string, meta schema.GraphMeta) (_ *schema.Graph, err error) { + ctx, endSpan := otel.StartSpan(manager.tracer, ctx, "RegisterGraph", + attribute.String("name", name), + attribute.String("meta", types.Stringify(meta)), + ) + defer func() { endSpan(err) }() + + var graph schema.Graph + if err := manager.queue.Tx(ctx, func(conn pg.Conn) error { + return conn.With("name", name, "version", manager.version).Insert(ctx, &graph, meta) + }); err != nil { + return nil, err + } + + // Return success + return types.Ptr(graph), nil +} diff --git a/pggraph/manager/manager.go b/pggraph/manager/manager.go new file mode 100644 index 0000000..2b3b862 --- /dev/null +++ b/pggraph/manager/manager.go @@ -0,0 +1,102 @@ +package manager + +import ( + "context" + "fmt" + "strings" + + // Packages + otel "github.com/mutablelogic/go-client/pkg/otel" + pg "github.com/mutablelogic/go-pg" + schema "github.com/mutablelogic/go-pg/pggraph/schema" + pgqueue "github.com/mutablelogic/go-pg/pgqueue/manager" + attribute "go.opentelemetry.io/otel/attribute" +) + +/////////////////////////////////////////////////////////////////////////////// +// TYPES + +type Manager struct { + opt + queue *pgqueue.Manager +} + +/////////////////////////////////////////////////////////////////////////////// +// LIFECYCLE + +func New(ctx context.Context, pool pg.PoolConn, version string, opts ...Opt) (*Manager, error) { + self := new(Manager) + + // Check arguments + if pool == nil { + return nil, fmt.Errorf("pool is required") + } + + // Set default values + if err := self.defaults(version); err != nil { + return nil, err + } + + // Apply options + if err := self.apply(opts...); err != nil { + return nil, err + } + + // Parse and register named queries so bind.Query(...) can resolve them. + queries, err := pg.NewQueries(strings.NewReader(schema.Queries)) + if err != nil { + return nil, fmt.Errorf("parse queries.sql: %w", err) + } else { + pool = pool.WithQueries(queries).With( + "schema", self.schema, + ).(pg.PoolConn) + } + + // Create objects in the database schema. This is not done in a transaction + bootstrapCtx, endBootstrapSpan := otel.StartSpan(self.tracer, ctx, "bootstrap", + attribute.String("schema", self.schema), + ) + if err := bootstrap(bootstrapCtx, pool, self.schema); err != nil { + endBootstrapSpan(err) + return nil, err + } + + // Create the queue manager + queue, err := pgqueue.New(ctx, pool, pgqueue.WithSchema(self.schema), pgqueue.WithTracer(self.tracer), pgqueue.WithMeter(self.metrics), pgqueue.WithWorker(self.worker)) + if err != nil { + endBootstrapSpan(err) + return nil, fmt.Errorf("create queue manager: %w", err) + } else { + self.queue = queue + } + + // Return success + endBootstrapSpan(nil) + return self, nil +} + +/////////////////////////////////////////////////////////////////////////////// +// PRIVATE METHODS + +func bootstrap(ctx context.Context, conn pg.Conn, schemaName string) error { + // Get all objects + objects, err := pg.NewQueries(strings.NewReader(schema.Objects)) + if err != nil { + return fmt.Errorf("parse objects.sql: %w", err) + } + + // Create the schema + if err := pg.SchemaCreate(ctx, conn, schemaName); err != nil { + return fmt.Errorf("create schema %q: %w", schemaName, err) + } + + // Create all objects - not in a transaction + for _, key := range objects.Keys() { + if err := conn.Exec(ctx, objects.Query(key)); err != nil { + return fmt.Errorf("create object %q: %w", key, err) + } + } + + // Return success + return nil +} diff --git a/pggraph/manager/node.go b/pggraph/manager/node.go new file mode 100644 index 0000000..7eb23ef --- /dev/null +++ b/pggraph/manager/node.go @@ -0,0 +1,118 @@ +package manager + +import ( + "context" + "fmt" + "reflect" + "runtime" + "strings" + + // Packages + otel "github.com/mutablelogic/go-client/pkg/otel" + pg "github.com/mutablelogic/go-pg" + schema "github.com/mutablelogic/go-pg/pggraph/schema" + queueschema "github.com/mutablelogic/go-pg/pgqueue/schema" + types "github.com/mutablelogic/go-server/pkg/types" + attribute "go.opentelemetry.io/otel/attribute" +) + +/////////////////////////////////////////////////////////////////////////////// +// PUBLIC METHODS + +// RegisterNode creates a new node, or updates an existing node, and returns it. +func (manager *Manager) RegisterNode(ctx context.Context, name string, meta schema.NodeMeta, fn any) (_ *schema.Node, err error) { + ctx, endSpan := otel.StartSpan(manager.tracer, ctx, "RegisterNode", + attribute.String("name", name), + attribute.String("meta", types.Stringify(meta)), + ) + defer func() { endSpan(err) }() + + // Determine the input and output edges from the function signature + _, in, out, err := functionEdges(fn) + if err != nil { + return nil, err + } + + // Create the node in the database + insert := schema.NodeInsert{ + Name: name, + In: out, + Out: in, + NodeMeta: meta, + } + + var node schema.Node + if err := manager.queue.Tx(ctx, func(conn pg.Conn) error { + // Create the node + if err := conn.With("version", manager.version).Insert(ctx, &node, insert); err != nil { + return err + } + + // TODO: Create the task queue for the node + _, err := manager.queue.RegisterQueue(ctx, name, queueschema.QueueMeta{}, manager.taskRunner(node, fn)) + if err != nil { + return err + } + + // Return success + return nil + }); err != nil { + return nil, err + } + + // Return the node + return types.Ptr(node), nil +} + +/////////////////////////////////////////////////////////////////////////////// +// PRIVATE METHODS + +func functionEdges(fn any) (string, []schema.Edge, schema.Edge, error) { + if fn == nil { + return "", nil, "", pg.ErrBadParameter.With("fn is missing") + } + + value := reflect.ValueOf(fn) + if value.Kind() != reflect.Func { + return "", nil, "", pg.ErrBadParameter.With("fn is not a function") + } + + fnType := value.Type() + if fnType.NumIn() < 2 { + return "", nil, "", pg.ErrBadParameter.With("fn must accept context.Context and at least one input") + } + if fnType.In(0) != reflect.TypeOf((*context.Context)(nil)).Elem() { + return "", nil, "", pg.ErrBadParameter.With("fn must accept context.Context as the first argument") + } + if fnType.NumOut() != 2 { + return "", nil, "", pg.ErrBadParameter.With("fn must return exactly one value and error") + } + if fnType.Out(1) != reflect.TypeOf((*error)(nil)).Elem() { + return "", nil, "", pg.ErrBadParameter.With("fn must return error as the last return value") + } + + runtimeFn := runtime.FuncForPC(value.Pointer()) + if runtimeFn == nil { + return "", nil, "", pg.ErrBadParameter.With("fn name is missing") + } + fnName := runtimeFn.Name() + if strings.TrimSpace(fnName) == "" { + return "", nil, "", pg.ErrBadParameter.With("fn name is missing") + } + + in := make([]schema.Edge, 0, fnType.NumIn()-1) + for i := 1; i < fnType.NumIn(); i++ { + edge := schema.Edge(fnType.In(i).String()) + if strings.TrimSpace(string(edge)) == "" { + return "", nil, "", fmt.Errorf("fn %q has an empty input type", fnName) + } + in = append(in, edge) + } + + out := schema.Edge(fnType.Out(0).String()) + if strings.TrimSpace(string(out)) == "" { + return "", nil, "", fmt.Errorf("fn %q has an empty return type", fnName) + } + + return fnName, in, out, nil +} diff --git a/pggraph/manager/opt.go b/pggraph/manager/opt.go new file mode 100644 index 0000000..872e93c --- /dev/null +++ b/pggraph/manager/opt.go @@ -0,0 +1,109 @@ +package manager + +import ( + "fmt" + "os" + "regexp" + + // Packages + schema "github.com/mutablelogic/go-pg/pgqueue/schema" + metric "go.opentelemetry.io/otel/metric" + trace "go.opentelemetry.io/otel/trace" +) + +/////////////////////////////////////////////////////////////////////////////// +// TYPES + +// Opt configures a Manager during construction. +type Opt func(*opt) error + +// opt combines all configuration options for Manager. +type opt struct { + worker string + schema string + version string + tracer trace.Tracer + metrics metric.Meter +} + +/////////////////////////////////////////////////////////////////////////////// +// GLOBALS + +var ( + reVersion = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$`) +) + +/////////////////////////////////////////////////////////////////////////////// +// LIFECYCLE + +func (o *opt) apply(opts ...Opt) error { + for _, opt := range opts { + if opt == nil { + continue + } + if err := opt(o); err != nil { + return err + } + } + return nil +} + +func (o *opt) defaults(version string) error { + o.schema = schema.DefaultSchema + + // Set version + if reVersion.MatchString(version) { + o.version = version + } else { + return fmt.Errorf("invalid version: %q", version) + } + + // Set worker + if hostname, err := os.Hostname(); err != nil { + return err + } else { + o.worker = hostname + } + + // Return success + return nil +} + +/////////////////////////////////////////////////////////////////////////////// +// PUBLIC METHODS + +// WithSchema sets the database schema names to use for all queries. If not set the default schemas are used. +func WithSchema(schemaName string) Opt { + return func(o *opt) error { + if schemaName != "" { + o.schema = schemaName + } + return nil + } +} + +// WithWorker sets the worker name used for manager tasks. If not set the hostname is used. +func WithWorker(workerName string) Opt { + return func(o *opt) error { + if workerName != "" { + o.worker = workerName + } + return nil + } +} + +// WithTracer sets the OpenTelemetry tracer used for manager spans. +func WithTracer(tracer trace.Tracer) Opt { + return func(o *opt) error { + o.tracer = tracer + return nil + } +} + +// WithMeter sets the OpenTelemetry meter used for manager metrics. +func WithMeter(meter metric.Meter) Opt { + return func(o *opt) error { + o.metrics = meter + return nil + } +} diff --git a/pggraph/manager/run.go b/pggraph/manager/run.go new file mode 100644 index 0000000..b9c765b --- /dev/null +++ b/pggraph/manager/run.go @@ -0,0 +1,14 @@ +package manager + +import ( + "context" + "log/slog" +) + +/////////////////////////////////////////////////////////////////////////////// +// PUBLIC METHODS + +func (manager *Manager) Run(ctx context.Context, log *slog.Logger) error { + <-ctx.Done() + return nil +} diff --git a/pggraph/schema/graph.go b/pggraph/schema/graph.go new file mode 100644 index 0000000..7715054 --- /dev/null +++ b/pggraph/schema/graph.go @@ -0,0 +1,83 @@ +package schema + +import ( + "strings" + "time" + + // Packages + pg "github.com/mutablelogic/go-pg" + types "github.com/mutablelogic/go-server/pkg/types" +) + +//////////////////////////////////////////////////////////////////////////////// +// TYPES + +type GraphName string + +type GraphMeta struct { + Description *string `json:"description,omitempty" help:"Human-readable description of the graph"` +} + +type Graph struct { + Name string `json:"graph" arg:"" help:"Graph name"` + Version string `json:"version" help:"Graph version"` + CreatedAt time.Time `json:"created_at" help:"When the graph was created"` + GraphMeta +} + +type GraphListRequest struct { + Version *string `json:"version,omitempty" help:"Graph version to filter by"` + pg.OffsetLimit +} + +type GraphList struct { + GraphListRequest + Count uint64 `json:"count"` + Body []*Graph `json:"body,omitempty"` +} + +//////////////////////////////////////////////////////////////////////////////// +// READER + +func (g *Graph) Scan(row pg.Row) error { + return row.Scan(&g.Name, &g.Version, &g.Description, &g.CreatedAt) +} + +//////////////////////////////////////////////////////////////////////////////// +// SELECTOR + +func (g GraphName) Select(bind *pg.Bind, op pg.Op) (string, error) { + if name := strings.TrimSpace(string(g)); !types.IsIdentifier(name) { + return "", pg.ErrBadParameter.With("graph name is not a valid identifier") + } else { + bind.Set("name", name) + } + if !bind.Has("version") { + return "", pg.ErrBadParameter.With("version is missing") + } + + switch op { + case pg.Get: + return bind.Query("pggraph.get"), nil + default: + return "", pg.ErrNotImplemented.Withf("unsupported GraphName operation %q", op) + } +} + +//////////////////////////////////////////////////////////////////////////////// +// WRITER + +func (g GraphMeta) Insert(bind *pg.Bind) (string, error) { + if !bind.Has("name") { + return "", pg.ErrBadParameter.With("name is missing") + } + if !bind.Has("version") { + return "", pg.ErrBadParameter.With("version is missing") + } + bind.Set("description", g.Description) + return bind.Query("pggraph.upsert"), nil +} + +func (g GraphMeta) Update(bind *pg.Bind) error { + return pg.ErrNotImplemented.With("GraphMeta does not support updates") +} diff --git a/pggraph/schema/node.go b/pggraph/schema/node.go new file mode 100644 index 0000000..682c553 --- /dev/null +++ b/pggraph/schema/node.go @@ -0,0 +1,95 @@ +package schema + +import ( + "strings" + + // Packages + pg "github.com/mutablelogic/go-pg" + types "github.com/mutablelogic/go-server/pkg/types" +) + +//////////////////////////////////////////////////////////////////////////////// +// TYPES + +type NodeName string + +type NodeMeta struct { + Description *string `json:"description,omitempty" help:"Human-readable description"` +} + +type NodeInsert struct { + Name string `json:"name" arg:"" help:"Node name"` + In Edge `json:"in,omitempty" help:"Incoming edge to this node"` + Out []Edge `json:"out,omitempty" help:"Outgoing edges from this node"` + NodeMeta +} + +type Node struct { + NodeInsert + Version string `json:"version" help:"Node version"` + RetainCount uint64 `json:"retain_count" help:"Retained node value"` +} + +type Edge string + +//////////////////////////////////////////////////////////////////////////////// +// READER + +func (n *Node) Scan(row pg.Row) error { + return row.Scan(&n.Name, &n.Version, &n.Description, &n.RetainCount) +} + +//////////////////////////////////////////////////////////////////////////////// +// SELECTOR + +func (n NodeName) Select(bind *pg.Bind, op pg.Op) (string, error) { + if name := strings.TrimSpace(string(n)); !types.IsIdentifier(name) { + return "", pg.ErrBadParameter.With("node name is not a valid identifier") + } else { + bind.Set("name", name) + } + if !bind.Has("version") { + return "", pg.ErrBadParameter.With("version is missing") + } + + switch op { + case pg.Get: + return bind.Query("pggraph.node_get"), nil + default: + return "", pg.ErrNotImplemented.Withf("unsupported NodeName operation %q", op) + } +} + +//////////////////////////////////////////////////////////////////////////////// +// WRITER + +func (n NodeInsert) Insert(bind *pg.Bind) (string, error) { + if name := strings.TrimSpace(n.Name); !types.IsIdentifier(name) { + return "", pg.ErrBadParameter.With("name is missing or not a valid identifier") + } else { + bind.Set("name", name) + } + if in := strings.TrimSpace(string(n.In)); in == "" { + return "", pg.ErrBadParameter.Withf("node %q requires one incoming edge", n.Name) + } else { + bind.Set("in", in) + } + if len(n.Out) == 0 { + return "", pg.ErrBadParameter.Withf("node %q requires one or more outgoing edges", n.Name) + } + out := make([]string, 0, len(n.Out)) + for _, edge := range n.Out { + if edgeType := strings.TrimSpace(string(edge)); edgeType == "" { + return "", pg.ErrBadParameter.Withf("node %q requires one or more outgoing edges", n.Name) + } else { + out = append(out, edgeType) + } + } + bind.Set("out", out) + bind.Set("description", n.Description) + return bind.Query("pggraph.node_upsert"), nil +} + +func (n NodeMeta) Update(bind *pg.Bind) error { + return nil +} diff --git a/pggraph/schema/objects.sql b/pggraph/schema/objects.sql new file mode 100644 index 0000000..e322df9 --- /dev/null +++ b/pggraph/schema/objects.sql @@ -0,0 +1,30 @@ +-- pggraph.graph +CREATE TABLE IF NOT EXISTS ${"schema"}."graph" ( + "name" TEXT NOT NULL CHECK ("name" ~ '^[A-Za-z_][A-Za-z0-9_]*$'), + "version" TEXT NOT NULL CHECK ("version" ~ '^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$'), + "description" TEXT, + "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY ("name", "version") +); + +-- pggraph.edge_type +DO $$ BEGIN + CREATE TYPE ${"schema"}."edge" AS ( + "type" TEXT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +-- pggraph.node +CREATE TABLE IF NOT EXISTS ${"schema"}."node" ( + "id" BIGINT GENERATED ALWAYS AS IDENTITY UNIQUE, + "name" TEXT NOT NULL CHECK ("name" ~ '^[A-Za-z_][A-Za-z0-9_]*$'), + "version" TEXT NOT NULL CHECK ("version" ~ '^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$'), + "description" TEXT, + "in" ${"schema"}."edge" NOT NULL, + "out" ${"schema"}."edge"[] NOT NULL DEFAULT ARRAY[]::${"schema"}."edge"[], + "retain_count" BIGINT NOT NULL DEFAULT 0 CHECK ("retain_count" >= 0), + CHECK (CARDINALITY("out") >= 1), + PRIMARY KEY ("name", "version") +); diff --git a/pggraph/schema/queries.sql b/pggraph/schema/queries.sql new file mode 100644 index 0000000..e0ec780 --- /dev/null +++ b/pggraph/schema/queries.sql @@ -0,0 +1,72 @@ +-- pggraph.get +SELECT + "name", + "version", + "description", + "created_at" +FROM + ${"schema"}."graph" +WHERE + "name" = @id +AND + "version" = @version; + +-- pggraph.upsert +INSERT INTO ${"schema"}."graph" ( + "name", + "version", + "description" +) VALUES ( + @id, + @version, + @description +) +ON CONFLICT ("name", "version") +DO UPDATE SET + "description" = EXCLUDED."description" +RETURNING + "name", + "version", + "description", + "created_at"; + +-- pggraph.node_get +SELECT + "name", + "version", + "description", + "retain_count" +FROM + ${"schema"}."node" +WHERE + "name" = @name +AND + "version" = @version; + +-- pggraph.node_upsert +INSERT INTO ${"schema"}."node" ( + "name", + "version", + "description", + "in", + "out" +) VALUES ( + @name, + @version, + @description, + ROW(@in)::${"schema"}."edge", + ARRAY( + SELECT ROW(v)::${"schema"}."edge" + FROM UNNEST(CAST(@out AS TEXT[])) AS v + ) +) +ON CONFLICT ("name", "version") +DO UPDATE SET + "description" = EXCLUDED."description", + "in" = EXCLUDED."in", + "out" = EXCLUDED."out" +RETURNING + "name", + "version", + "description", + "retain_count"; diff --git a/pggraph/schema/schema.go b/pggraph/schema/schema.go new file mode 100644 index 0000000..ee4f49b --- /dev/null +++ b/pggraph/schema/schema.go @@ -0,0 +1,18 @@ +package schema + +import ( + _ "embed" +) + +/////////////////////////////////////////////////////////////////////////////// +// GLOBALS + +//go:embed objects.sql +var Objects string + +//go:embed queries.sql +var Queries string + +const ( + DefaultSchema = "pggraph" +)