8 files changed, 191 insertions(+), 43 deletions(-)

A api/account/middleware.go
M api/graph/schema.graphqls
M api/graph/schema.resolvers.go
M api/repos/middleware.go
M api/server.go
M gitsrht-dispatch/main.go
M gitsrht-keys/main.go
M schema.sql

@@ 0,0 1,110 @@
+package account
+
+import (
+	"context"
+	"database/sql"
+	"log"
+	"net/http"
+	"os"
+	"path"
+
+	"git.sr.ht/~sircmpwn/core-go/config"
+	"git.sr.ht/~sircmpwn/core-go/database"
+	work "git.sr.ht/~sircmpwn/dowork"
+	"git.sr.ht/~sircmpwn/git.sr.ht/api/repos"
+)
+
+type contextKey struct {
+	name string
+}
+
+var ctxKey = &contextKey{"account"}
+
+func Middleware(queue *work.Queue) func(next http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			ctx := context.WithValue(r.Context(), ctxKey, queue)
+			r = r.WithContext(ctx)
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
+// Schedules a user account deletion.
+func Delete(ctx context.Context, userID int, username string) {
+	queue, ok := ctx.Value(ctxKey).(*work.Queue)
+	if !ok {
+		panic("No account worker for this context")
+	}
+
+	type Artifact struct {
+		Filename string
+		RepoName string
+	}
+
+	conf := config.ForContext(ctx)
+	repoStore, ok := conf.Get("git.sr.ht", "repos")
+
+	task := work.NewTask(func(ctx context.Context) error {
+		log.Printf("Processing deletion of user account %d %s", userID, username)
+		var artifacts []Artifact
+		if err := database.WithTx(ctx, &sql.TxOptions{
+			Isolation: 0,
+			ReadOnly:  true,
+		}, func(tx *sql.Tx) error {
+			rows, err := tx.QueryContext(ctx, `
+				SELECT r.name, a.filename
+				FROM artifacts a
+				JOIN repository r ON a.repo_id = r.id
+				WHERE a.user_id = $1
+			`, userID)
+			if err != nil {
+				return err
+			}
+
+			for rows.Next() {
+				var (
+					filename string
+					repoName string
+				)
+				if err := rows.Scan(&repoName, &filename); err != nil {
+					return err
+				}
+				artifacts = append(artifacts, Artifact{
+					Filename: filename,
+					RepoName: repoName,
+				})
+			}
+			if err := rows.Err(); err != nil {
+				return err
+			}
+
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		for _, art := range artifacts {
+			repos.DeleteArtifactsBlocking(ctx, username,
+				art.RepoName, []string{art.Filename})
+		}
+		userPath := path.Join(repoStore, "~"+username)
+		if err := os.RemoveAll(userPath); err != nil {
+			log.Printf("Failed to remove %s: %s", userPath, err.Error())
+		}
+
+		if err := database.WithTx(ctx, nil, func(tx *sql.Tx) error {
+			_, err := tx.ExecContext(ctx, `
+				DELETE FROM "user" WHERE id = $1
+			`, userID)
+			return err
+		}); err != nil {
+			return err
+		}
+
+		log.Printf("Deletion of user account %d %s complete", userID, username)
+		return nil
+	})
+	queue.Enqueue(task)
+	log.Printf("Enqueued deletion of user account %d %s", userID, username)
+}

@@ 13,6 13,12 @@ access token, and are not available to clients using OAuth 2.0 access tokens.
 """
 directive @private on FIELD_DEFINITION
 
+"""
+This used to decorate fields which are for internal use, and are not
+available to normal API users.
+"""
+directive @internal on FIELD_DEFINITION
+
 enum AccessScope {
   PROFILE      @scopehelp(details: "profile information")
   REPOSITORIES @scopehelp(details: "repository metadata")


@@ 567,4 573,9 @@ type Mutation {
   unexpected behavior with the third-party integration.
   """
   deleteWebhook(id: Int!): WebhookSubscription
+
+  """
+  Deletes the authenticated user's account. Internal use only.
+  """
+  deleteUser: Int! @internal
 }

@@ 28,6 28,7 @@ import (
 	"git.sr.ht/~sircmpwn/core-go/server"
 	"git.sr.ht/~sircmpwn/core-go/valid"
 	corewebhooks "git.sr.ht/~sircmpwn/core-go/webhooks"
+	"git.sr.ht/~sircmpwn/git.sr.ht/api/account"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/graph/api"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/graph/model"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/loaders"


@@ 857,6 858,13 @@ func (r *mutationResolver) DeleteWebhook(ctx context.Context, id int) (model.Web
 	return &sub, nil
 }
 
+// DeleteUser is the resolver for the deleteUser field.
+func (r *mutationResolver) DeleteUser(ctx context.Context) (int, error) {
+	user := auth.ForContext(ctx)
+	account.Delete(ctx, user.UserID, user.Username)
+	return user.UserID, nil
+}
+
 // Version is the resolver for the version field.
 func (r *queryResolver) Version(ctx context.Context) (*model.Version, error) {
 	conf := config.ForContext(ctx)

@@ 49,7 49,8 @@ func Clone(ctx context.Context, repoID int, repo *git.Repository, cloneURL strin
 		panic("No repos worker for this context")
 	}
 	task := work.NewTask(func(ctx context.Context) error {
-		cloneCtx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+		log.Printf("Processing clone of %s", cloneURL)
+		cloneCtx, cancel := context.WithTimeout(ctx, 30*time.Minute)
 		defer cancel()
 		err := repo.Clone(cloneCtx, &git.CloneOptions{
 			URL:               cloneURL,


@@ 74,6 75,7 @@ func Clone(ctx context.Context, repoID int, repo *git.Repository, cloneURL strin
 		}); err != nil {
 			panic(err)
 		}
+		log.Printf("Clone %s complete", cloneURL)
 		return nil
 	})
 	queue.Enqueue(task)


@@ 87,33 89,42 @@ func DeleteArtifacts(ctx context.Context, username, repoName string, filenames [
 		panic("No repos worker for this context")
 	}
 	task := work.NewTask(func(ctx context.Context) error {
-		conf := config.ForContext(ctx)
-		upstream, _ := conf.Get("objects", "s3-upstream")
-		accessKey, _ := conf.Get("objects", "s3-access-key")
-		secretKey, _ := conf.Get("objects", "s3-secret-key")
-		bucket, _ := conf.Get("git.sr.ht", "s3-bucket")
-		prefix, _ := conf.Get("git.sr.ht", "s3-prefix")
+		return DeleteArtifactsBlocking(ctx, username, repoName, filenames)
+	})
+	queue.Enqueue(task)
+	log.Printf("Enqueued deletion of %d artifacts", len(filenames))
+}
 
-		if upstream == "" || accessKey == "" || secretKey == "" || bucket == "" {
-			return fmt.Errorf("Object storage is not enabled for this server")
-		}
+func DeleteArtifactsBlocking(
+	ctx context.Context,
+	username,
+	repoName string,
+	filenames []string,
+) error {
+	conf := config.ForContext(ctx)
+	upstream, _ := conf.Get("objects", "s3-upstream")
+	accessKey, _ := conf.Get("objects", "s3-access-key")
+	secretKey, _ := conf.Get("objects", "s3-secret-key")
+	bucket, _ := conf.Get("git.sr.ht", "s3-bucket")
+	prefix, _ := conf.Get("git.sr.ht", "s3-prefix")
 
-		mc, err := minio.New(upstream, &minio.Options{
-			Creds:  credentials.NewStaticV4(accessKey, secretKey, ""),
-			Secure: true,
-		})
-		if err != nil {
-			panic(err)
-		}
+	if upstream == "" || accessKey == "" || secretKey == "" || bucket == "" {
+		return fmt.Errorf("Object storage is not enabled for this server")
+	}
 
-		for _, filename := range filenames {
-			s3path := path.Join(prefix, "artifacts", "~"+username, repoName, filename)
-			if err := mc.RemoveObject(ctx, bucket, s3path, minio.RemoveObjectOptions{}); err != nil {
-				return err
-			}
-		}
-		return nil
+	mc, err := minio.New(upstream, &minio.Options{
+		Creds:  credentials.NewStaticV4(accessKey, secretKey, ""),
+		Secure: true,
 	})
-	queue.Enqueue(task)
-	log.Printf("Enqueued deletion of %d artifacts", len(filenames))
+	if err != nil {
+		panic(err)
+	}
+
+	for _, filename := range filenames {
+		s3path := path.Join(prefix, "artifacts", "~"+username, repoName, filename)
+		if err := mc.RemoveObject(ctx, bucket, s3path, minio.RemoveObjectOptions{}); err != nil {
+			return err
+		}
+	}
+	return nil
 }

@@ 9,6 9,7 @@ import (
 	work "git.sr.ht/~sircmpwn/dowork"
 	"github.com/99designs/gqlgen/graphql"
 
+	"git.sr.ht/~sircmpwn/git.sr.ht/api/account"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/graph"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/graph/api"
 	"git.sr.ht/~sircmpwn/git.sr.ht/api/graph/model"


@@ 21,6 22,7 @@ func main() {
 
 	gqlConfig := api.Config{Resolvers: &graph.Resolver{}}
 	gqlConfig.Directives.Private = server.Private
+	gqlConfig.Directives.Internal = server.Internal
 	gqlConfig.Directives.Access = func(ctx context.Context, obj interface{},
 		next graphql.Resolver, scope model.AccessScope,
 		kind model.AccessKind) (interface{}, error) {


@@ 34,6 36,7 @@ func main() {
 	}
 
 	reposQueue := work.NewQueue("repos")
+	accountQueue := work.NewQueue("account")
 	webhookQueue := webhooks.NewQueue(schema)
 	legacyWebhooks := webhooks.NewLegacyQueue()
 


@@ 41,11 44,16 @@ func main() {
 		WithDefaultMiddleware().
 		WithMiddleware(
 			loaders.Middleware,
+			account.Middleware(accountQueue),
 			repos.Middleware(reposQueue),
 			webhooks.Middleware(webhookQueue),
 			webhooks.LegacyMiddleware(legacyWebhooks),
 		).
 		WithSchema(schema, scopes).
-		WithQueues(reposQueue, webhookQueue.Queue, legacyWebhooks.Queue).
+		WithQueues(
+			accountQueue,
+			reposQueue,
+			webhookQueue.Queue,
+			legacyWebhooks.Queue).
 		Run()
 }

@@ 2,8 2,8 @@ package main
 
 import (
 	"fmt"
-	"log"
 	"io"
+	"log"
 	"os"
 	osuser "os/user"
 	"strconv"


@@ 14,9 14,9 @@ import (
 )
 
 type Dispatcher struct {
-	cmd string
-	uid int
-	gid int
+	cmd  string
+	uid  int
+	gid  int
 	gids []int
 }
 


@@ 30,7 30,7 @@ func main() {
 	logf, err := os.OpenFile("/var/log/gitsrht-dispatch",
 		os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
 	if err != nil {
-		log.Printf("Warning: unable to open log file: %v " +
+		log.Printf("Warning: unable to open log file: %v "+
 			"(using stderr instead)", err)
 		logger = log.New(os.Stderr, "", log.LstdFlags)
 	} else {

@@ 5,9 5,9 @@ import (
 	"os"
 	"path"
 
+	"git.sr.ht/~sircmpwn/scm.sr.ht/srht-keys"
 	goredis "github.com/go-redis/redis/v8"
 	"github.com/vaughan0/go-ini"
-	"git.sr.ht/~sircmpwn/scm.sr.ht/srht-keys"
 )
 
 func main() {

@@ 61,7 61,7 @@ CREATE TABLE repository (
 	updated timestamp without time zone NOT NULL,
 	name character varying(256) NOT NULL,
 	description character varying(1024),
-	owner_id integer NOT NULL REFERENCES "user"(id),
+	owner_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
 	path character varying(1024),
 	visibility visibility NOT NULL,
 	readme character varying,


@@ 77,7 77,7 @@ CREATE TABLE access (
 	created timestamp without time zone NOT NULL,
 	updated timestamp without time zone NOT NULL,
 	repo_id integer NOT NULL REFERENCES repository(id) ON DELETE CASCADE,
-	user_id integer NOT NULL REFERENCES "user"(id),
+	user_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
 	mode character varying NOT NULL,
 	CONSTRAINT uq_access_user_id_repo_id UNIQUE (user_id, repo_id)
 );


@@ 85,8 85,8 @@ CREATE TABLE access (
 CREATE TABLE artifacts (
 	id serial PRIMARY KEY,
 	created timestamp without time zone NOT NULL,
-	user_id integer NOT NULL REFERENCES "user"(id),
-	repo_id integer NOT NULL REFERENCES repository(id),
+	user_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
+	repo_id integer NOT NULL REFERENCES repository(id) ON DELETE CASCADE,
 	commit character varying NOT NULL,
 	filename character varying NOT NULL,
 	checksum character varying NOT NULL,


@@ 98,7 98,7 @@ CREATE TABLE redirect (
 	id serial PRIMARY KEY,
 	created timestamp without time zone NOT NULL,
 	name character varying(256) NOT NULL,
-	owner_id integer NOT NULL REFERENCES "user"(id),
+	owner_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
 	path character varying(1024),
 	new_repo_id integer NOT NULL REFERENCES repository(id) ON DELETE CASCADE
 );


@@ 116,7 116,7 @@ CREATE TABLE gql_user_wh_sub (
 	client_id uuid,
 	expires timestamp without time zone,
 	node_id character varying,
-	user_id integer NOT NULL REFERENCES "user"(id),
+	user_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
 	CONSTRAINT gql_user_wh_sub_auth_method_check
 		CHECK ((auth_method = ANY (ARRAY['OAUTH2'::auth_method, 'INTERNAL'::auth_method]))),
 	CONSTRAINT gql_user_wh_sub_check


@@ 149,7 149,7 @@ CREATE TABLE gql_user_wh_delivery (
 -- Legacy SSH key table, to be fetched from meta.sr.ht instead (TODO: Remove)
 CREATE TABLE sshkey (
 	id serial PRIMARY KEY,
-	user_id integer NOT NULL REFERENCES "user"(id),
+	user_id integer NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
 	meta_id integer NOT NULL,
 	key character varying(4096) NOT NULL,
 	fingerprint character varying(512) NOT NULL


@@ 169,7 169,7 @@ CREATE TABLE oauthtoken (
 	created timestamp without time zone NOT NULL,
 	updated timestamp without time zone NOT NULL,
 	expires timestamp without time zone NOT NULL,
-	user_id integer REFERENCES "user"(id),
+	user_id integer REFERENCES "user"(id) ON DELETE CASCADE,
 	token_hash character varying(128) NOT NULL,
 	token_partial character varying(8) NOT NULL,
 	scopes character varying(512) NOT NULL


@@ 181,8 181,8 @@ CREATE TABLE user_webhook_subscription (
 	created timestamp without time zone NOT NULL,
 	url character varying(2048) NOT NULL,
 	events character varying NOT NULL,
-	user_id integer REFERENCES "user"(id),
-	token_id integer REFERENCES oauthtoken(id)
+	user_id integer REFERENCES "user"(id) ON DELETE CASCADE,
+	token_id integer REFERENCES oauthtoken(id) ON DELETE CASCADE
 );
 
 CREATE TABLE user_webhook_delivery (