CRUD APIs and security policies for scheduled reports (#3293)

* Support ad-hoc code files backed by Postgres

* Typo

* Implement database functions

* Background job to garbage collect soft deleted virtual files

* Format query

* Self review

* CRUD APIs and security policies for scheduled reports

* Fix proto linter

* Fix golang lint

* Self review

* nits

* Add permissions for reports

* Incorporate permissions

* Fix migration

* QA on APIs

* Generate pretty report YAML

* Pretty export format

admin/database/database.go

@@ -562,6 +562,8 @@ type ProjectRole struct {
 	ManageDev            bool `db:"manage_dev"`
 	ReadProjectMembers   bool `db:"read_project_members"`
 	ManageProjectMembers bool `db:"manage_project_members"`
+	CreateReports        bool `db:"create_reports"`
+	ManageReports        bool `db:"manage_reports"`
 }
 
 // Member is a convenience type used for display-friendly representation of an org or project member.

admin/database/postgres/migrations/0018.sql

@@ -0,0 +1,5 @@
+ALTER TABLE project_roles ADD create_reports BOOLEAN NOT NULL DEFAULT false;
+UPDATE project_roles SET create_reports = read_prod;
+
+ALTER TABLE project_roles ADD manage_reports BOOLEAN NOT NULL DEFAULT false;
+UPDATE project_roles SET manage_reports = manage_prod;

admin/database/postgres/postgres.go

@@ -1268,7 +1268,7 @@ func (c *connection) FindVirtualFiles(ctx context.Context, projectID, branch str
 	err := c.getDB(ctx).SelectContext(ctx, &res, `
 		SELECT path, data, deleted, updated_on
 		FROM virtual_files
-		WHERE project_id=$1 AND branch=$2 AND (updated_on>$3 OR updated_on=$3 AND afterPath>$4)
+		WHERE project_id=$1 AND branch=$2 AND (updated_on>$3 OR updated_on=$3 AND path>$4)
 		ORDER BY updated_on, path LIMIT $5
 	`, projectID, branch, afterUpdatedOn, afterPath, limit)
 	if err != nil {
@@ -1302,7 +1302,7 @@ func (c *connection) UpdateVirtualFileDeleted(ctx context.Context, projectID, br
 			data = ''::BYTEA,
 			deleted = TRUE,
 			updated_on = now()
-		WHERE project_id=$1, branch=$2, path=$3`, projectID, branch, path)
+		WHERE project_id=$1 AND branch=$2 AND path=$3`, projectID, branch, path)
 	return checkUpdateRow("virtual file", res, err)
 }
 

admin/deployments.go

@@ -135,7 +135,7 @@ func (s *Service) createDeployment(ctx context.Context, opts *createDeploymentOp
 				Name: "admin",
 				Type: "admin",
 				Config: map[string]string{
-					"host":         s.opts.ExternalURL,
+					"admin_url":    s.opts.ExternalURL,
 					"access_token": adminAuthToken,
 					"project_id":   opts.ProjectID,
 					"branch":       opts.ProdBranch,

admin/permissions.go

@@ -67,6 +67,8 @@ func (s *Service) ProjectPermissionsForUser(ctx context.Context, projectID, user
 			ManageDev:            true,
 			ReadProjectMembers:   true,
 			ManageProjectMembers: true,
+			CreateReports:        true,
+			ManageReports:        true,
 		}, nil
 	}
 
@@ -98,6 +100,8 @@ func (s *Service) ProjectPermissionsForService(ctx context.Context, projectID, s
 			ManageDev:            true,
 			ReadProjectMembers:   true,
 			ManageProjectMembers: true,
+			CreateReports:        true,
+			ManageReports:        true,
 		}, nil
 	}
 
@@ -125,6 +129,8 @@ func (s *Service) ProjectPermissionsForDeployment(ctx context.Context, projectID
 			ManageDev:            false,
 			ReadProjectMembers:   true,
 			ManageProjectMembers: false,
+			CreateReports:        false,
+			ManageReports:        false,
 		}, nil
 	}
 
@@ -155,5 +161,7 @@ func unionProjectRoles(a *adminv1.ProjectPermissions, b *database.ProjectRole) *
 		ManageDev:            a.ManageDev || b.ManageDev,
 		ReadProjectMembers:   a.ReadProjectMembers || b.ReadProjectMembers,
 		ManageProjectMembers: a.ManageProjectMembers || b.ManageProjectMembers,
+		CreateReports:        a.CreateReports || b.CreateReports,
+		ManageReports:        a.ManageReports || b.ManageReports,
 	}
 }

admin/reports.go

@@ -0,0 +1,128 @@
+package admin
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/rilldata/rill/admin/database"
+	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
+	"github.com/rilldata/rill/runtime"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// TriggerReport triggers an ad-hoc run of a report
+func (s *Service) TriggerReport(ctx context.Context, depl *database.Deployment, report string) (err error) {
+	names := []*runtimev1.ResourceName{
+		{
+			Kind: runtime.ResourceKindReport,
+			Name: report,
+		},
+	}
+
+	rt, err := s.openRuntimeClientForDeployment(depl)
+	if err != nil {
+		return err
+	}
+	defer rt.Close()
+
+	_, err = rt.CreateTrigger(ctx, &runtimev1.CreateTriggerRequest{
+		InstanceId: depl.RuntimeInstanceID,
+		Trigger: &runtimev1.CreateTriggerRequest_RefreshTriggerSpec{
+			RefreshTriggerSpec: &runtimev1.RefreshTriggerSpec{OnlyNames: names},
+		},
+	})
+	return err
+}
+
+// TriggerReconcileAndAwaitReport triggers a reconcile and polls the runtime until the given report's spec version has been updated (or ctx is canceled).
+func (s *Service) TriggerReconcileAndAwaitReport(ctx context.Context, depl *database.Deployment, reportName string) error {
+	rt, err := s.openRuntimeClientForDeployment(depl)
+	if err != nil {
+		return err
+	}
+	defer rt.Close()
+
+	reportReq := &runtimev1.GetResourceRequest{
+		InstanceId: depl.RuntimeInstanceID,
+		Name: &runtimev1.ResourceName{
+			Kind: runtime.ResourceKindReport,
+			Name: reportName,
+		},
+	}
+
+	// Get old spec version
+	var oldSpecVersion *int64
+	r, err := rt.GetResource(ctx, reportReq)
+	if err == nil {
+		oldSpecVersion = &r.Resource.Meta.SpecVersion
+	}
+
+	// Trigger reconcile
+	_, err = rt.CreateTrigger(ctx, &runtimev1.CreateTriggerRequest{
+		InstanceId: depl.RuntimeInstanceID,
+		Trigger: &runtimev1.CreateTriggerRequest_PullTriggerSpec{
+			PullTriggerSpec: &runtimev1.PullTriggerSpec{},
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	// Poll every 1 seconds until the report is found or the ctx is cancelled or times out
+	ctx, cancel := context.WithTimeout(ctx, time.Minute)
+	defer cancel()
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		r, err := rt.GetResource(ctx, reportReq)
+		if err != nil {
+			if s, ok := status.FromError(err); !ok || s.Code() != codes.NotFound {
+				return fmt.Errorf("failed to poll for report: %w", err)
+			}
+			if oldSpecVersion != nil {
+				// Success - previously the report was found, now we cannot find it anymore
+				return nil
+			}
+			// Continue polling
+			continue
+		}
+		if oldSpecVersion == nil {
+			// Success - previously the report was not found, now we found one
+			return nil
+		}
+		if *oldSpecVersion != r.Resource.Meta.SpecVersion {
+			// Success - the spec version has changed
+			return nil
+		}
+	}
+}
+
+// LookupReport fetches a report's spec from a runtime deployment.
+func (s *Service) LookupReport(ctx context.Context, depl *database.Deployment, reportName string) (*runtimev1.ReportSpec, error) {
+	rt, err := s.openRuntimeClientForDeployment(depl)
+	if err != nil {
+		return nil, err
+	}
+	defer rt.Close()
+
+	res, err := rt.GetResource(ctx, &runtimev1.GetResourceRequest{
+		InstanceId: depl.RuntimeInstanceID,
+		Name: &runtimev1.ResourceName{
+			Kind: runtime.ResourceKindReport,
+			Name: reportName,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return res.Resource.Resource.(*runtimev1.Resource_Report).Report.Spec, nil
+}