feat(scheduler,oauth-setup): Phase 3F — Scheduler + OAuth bootstrap helper

README.md

@@ -25,6 +25,66 @@ Run the API directly without Docker:
 make run
 ```
 
+## Real Gmail → Drive setup (one-time)
+
+If you want to actually run the Gmail-to-Drive sync (rather than just synthetic
+loadtest jobs), you need to set up a Google Cloud OAuth client and bootstrap a
+token. About 10 minutes of clicking.
+
+1. **Google Cloud Console:** Create (or reuse) a project. Enable the **Gmail API**
+   and **Drive API**. On the OAuth consent screen, set the scopes to
+   `gmail.readonly` and `drive.file`. Add your email as a test user.
+
+2. **OAuth client:** Create an OAuth 2.0 Client ID, type **Desktop**.
+   Add `http://localhost:8888/callback` as an authorized redirect URI.
+   Note the client ID and secret.
+
+3. **Encryption key:** Generate a 32-byte random key:
+
+   ```bash
+   openssl rand -base64 32
+   ```
+
+   Save it; the same value is needed by both `oauth-setup` and the workers/scheduler.
+
+4. **Bring up the local stack:**
+
+   ```bash
+   make docker-up
+   ```
+
+5. **Bootstrap your OAuth token:**
+
+   ```bash
+   export DATABASE_URL='postgres://dtq:dtq@localhost:5432/dtq?sslmode=disable'
+   export GOOGLE_OAUTH_CLIENT_ID='...'
+   export GOOGLE_OAUTH_CLIENT_SECRET='...'
+   export TOKEN_ENCRYPTION_KEY='<the base64 key>'
+   go run ./cmd/oauth-setup --email=you@example.com
+   ```
+
+   Open the printed URL in a browser, authorize the app, the helper exits with
+   "Token saved".
+
+6. **Run the workers + scheduler:**
+
+   ```bash
+   # in one terminal each, or via docker compose:
+   ./worker --stage=fetch
+   ./worker --stage=render
+   ./worker --stage=upload
+   ./scheduler
+   ```
+
+   On the first poll the scheduler initializes your cursor from your current
+   Gmail history ID (no backfill of existing inbox). Subsequent polls pick up
+   new INBOX/PRIMARY messages and push them through the fetch → render → upload
+   pipeline.
+
+7. **Where do the PDFs land?** A new folder tree at the root of your Drive:
+   `<root>/YYYY/MM/DD/`. Set `DRIVE_ROOT_FOLDER_ID` to the Drive folder ID where
+   you want this tree to live (or leave unset to land at the root).
+
 ## Tests
 
 ```bash

cmd/oauth-setup/main.go

@@ -0,0 +1,155 @@
+// Package main is a one-off helper to populate the oauth_tokens table with a
+// real Google OAuth token for a given email. Run once per user.
+//
+// Usage:
+//
+//	./oauth-setup --email=alice@example.com
+//
+// Environment variables required:
+//
+//	DATABASE_URL              postgres connection string
+//	GOOGLE_OAUTH_CLIENT_ID    from Google Cloud OAuth 2.0 Client IDs page
+//	GOOGLE_OAUTH_CLIENT_SECRET
+//	TOKEN_ENCRYPTION_KEY      base64 of 32 random bytes (also used by workers)
+//
+// The redirect URI configured in Google Cloud must include http://localhost:8888/callback.
+package main
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/google"
+
+	"github.com/smallchungus/disttaskqueue/internal/oauth"
+	"github.com/smallchungus/disttaskqueue/internal/store"
+)
+
+const callbackPort = "8888"
+
+func main() {
+	email := flag.String("email", "", "email address (must match the Google account you'll authorize)")
+	flag.Parse()
+	if *email == "" {
+		fmt.Fprintln(os.Stderr, "--email is required")
+		os.Exit(2)
+	}
+
+	dsn := mustEnv("DATABASE_URL")
+	clientID := mustEnv("GOOGLE_OAUTH_CLIENT_ID")
+	clientSecret := mustEnv("GOOGLE_OAUTH_CLIENT_SECRET")
+	keyB64 := mustEnv("TOKEN_ENCRYPTION_KEY")
+
+	key, err := base64.StdEncoding.DecodeString(keyB64)
+	if err != nil || len(key) != 32 {
+		log.Fatalf("TOKEN_ENCRYPTION_KEY must decode to 32 bytes: %v", err)
+	}
+
+	ctx := context.Background()
+	if err := store.Migrate(ctx, dsn); err != nil {
+		log.Fatalf("migrate: %v", err)
+	}
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		log.Fatalf("pg connect: %v", err)
+	}
+	defer pool.Close()
+	s := store.New(pool)
+
+	cfg := &oauth2.Config{
+		ClientID:     clientID,
+		ClientSecret: clientSecret,
+		RedirectURL:  "http://localhost:" + callbackPort + "/callback",
+		Scopes:       []string{"https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/drive.file"},
+		Endpoint:     google.Endpoint,
+	}
+
+	state := fmt.Sprintf("dtq-%d", time.Now().UnixNano())
+	authURL := cfg.AuthCodeURL(state, oauth2.AccessTypeOffline, oauth2.ApprovalForce)
+
+	fmt.Println("Open this URL in your browser:")
+	fmt.Println()
+	fmt.Println("  " + authURL)
+	fmt.Println()
+	fmt.Println("Authorize the app for " + *email + " and complete the redirect to localhost.")
+	fmt.Println()
+
+	codeCh := make(chan string, 1)
+	errCh := make(chan error, 1)
+	mux := http.NewServeMux()
+	srv := &http.Server{
+		Addr:              ":" + callbackPort,
+		Handler:           mux,
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+	mux.HandleFunc("/callback", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("state") != state {
+			errCh <- errors.New("state mismatch")
+			http.Error(w, "state mismatch", http.StatusBadRequest)
+			return
+		}
+		code := r.URL.Query().Get("code")
+		if code == "" {
+			errCh <- errors.New("no code in callback")
+			http.Error(w, "no code", http.StatusBadRequest)
+			return
+		}
+		_, _ = w.Write([]byte("OK. You can close this tab and return to the terminal."))
+		codeCh <- code
+	})
+	go func() {
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			errCh <- err
+		}
+	}()
+	defer func() { _ = srv.Shutdown(context.Background()) }()
+
+	var code string
+	select {
+	case code = <-codeCh:
+	case err := <-errCh:
+		log.Fatalf("callback: %v", err)
+	case <-time.After(5 * time.Minute):
+		log.Fatal("timed out waiting for callback")
+	}
+
+	tok, err := cfg.Exchange(ctx, code)
+	if err != nil {
+		log.Fatalf("exchange: %v", err)
+	}
+
+	user, err := s.GetUserByEmail(ctx, *email)
+	if errors.Is(err, store.ErrUserNotFound) {
+		user, err = s.CreateUser(ctx, *email)
+		if err != nil {
+			log.Fatalf("create user: %v", err)
+		}
+		fmt.Println("Created user:", user.ID)
+	} else if err != nil {
+		log.Fatalf("lookup user: %v", err)
+	} else {
+		fmt.Println("Existing user:", user.ID)
+	}
+
+	if err := oauth.SaveToken(ctx, s, user.ID, key, "google", tok); err != nil {
+		log.Fatalf("save token: %v", err)
+	}
+	fmt.Println("Token saved for user", user.ID, "(", *email, "). Scheduler can now sync this account.")
+}
+
+func mustEnv(k string) string {
+	v := os.Getenv(k)
+	if v == "" {
+		log.Fatalf("missing env var: %s", k)
+	}
+	return v
+}

cmd/scheduler/main.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"encoding/base64"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	goredis "github.com/redis/go-redis/v9"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/google"
+
+	"github.com/smallchungus/disttaskqueue/internal/queue"
+	"github.com/smallchungus/disttaskqueue/internal/scheduler"
+	"github.com/smallchungus/disttaskqueue/internal/store"
+)
+
+func main() {
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+	slog.SetDefault(logger)
+
+	dsn := envOr("DATABASE_URL", "postgres://dtq:dtq@localhost:5432/dtq?sslmode=disable")
+	redisURL := envOr("REDIS_URL", "redis://localhost:6379/0")
+	clientID := mustEnv("GOOGLE_OAUTH_CLIENT_ID")
+	clientSecret := mustEnv("GOOGLE_OAUTH_CLIENT_SECRET")
+	keyB64 := mustEnv("TOKEN_ENCRYPTION_KEY")
+
+	key, err := base64.StdEncoding.DecodeString(keyB64)
+	if err != nil || len(key) != 32 {
+		slog.Error("TOKEN_ENCRYPTION_KEY must decode to 32 bytes", "err", err)
+		os.Exit(1)
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	if err := store.Migrate(ctx, dsn); err != nil {
+		slog.Error("migrate", "err", err)
+		os.Exit(1)
+	}
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		slog.Error("pg connect", "err", err)
+		os.Exit(1)
+	}
+	defer pool.Close()
+
+	opts, err := goredis.ParseURL(redisURL)
+	if err != nil {
+		slog.Error("redis url", "err", err)
+		os.Exit(1)
+	}
+	redis := goredis.NewClient(opts)
+	defer func() { _ = redis.Close() }()
+
+	sch := scheduler.New(scheduler.Config{
+		Store:         store.New(pool),
+		Queue:         queue.New(redis),
+		EncryptionKey: key,
+		OAuth2: &oauth2.Config{
+			ClientID:     clientID,
+			ClientSecret: clientSecret,
+			Endpoint:     google.Endpoint,
+			Scopes:       []string{"https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/drive.file"},
+		},
+		Interval: 5 * time.Minute,
+	})
+
+	slog.Info("scheduler starting")
+	if err := sch.Run(ctx); err != nil {
+		slog.Error("scheduler run", "err", err)
+		os.Exit(1)
+	}
+	slog.Info("scheduler stopped")
+}
+
+func envOr(k, def string) string {
+	if v := os.Getenv(k); v != "" {
+		return v
+	}
+	return def
+}
+
+func mustEnv(k string) string {
+	v := os.Getenv(k)
+	if v == "" {
+		slog.Error("missing env var", "key", k)
+		os.Exit(1)
+	}
+	return v
+}